Hadoop中DataNode的啟動(dòng)過(guò)程詳解

2023年12月9日發(fā)(作者：致母親)

-

本文的目的在于詳細(xì)記錄2.4.1版本下hadoop的DataNode的啟動(dòng)過(guò)程，作此記錄，也為以后回過(guò)頭看DataNode留下方便。本文的思路是結(jié)合DataNode的代碼，來(lái)分析他的啟動(dòng)過(guò)程。

說(shuō)明，限于篇幅，對(duì)文章中引用的代碼都只留了關(guān)鍵部分。

DataNode同NameNode都一樣，是一個(gè)java進(jìn)程，所以從main方法開始，看代碼：

?

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

public class DataNode extends Configured

implements InterDatanodeProtocol, ClientDatanodeProtocol,

DataNodeMXBean {

public static void main(String args[]) {

cureMain(args, null);

}

public static void cureMain(String args[], SecureResources resources) {

DataNode datanode = createDataNode(args, null, resources);

}

public static DataNode createDataNode(String args[], Configuration conf,

SecureResources resources) throws IOException {

DataNode dn = instantiateDataNode(args, conf, resources);

if (dn != null) {

anodeDaemon();

}

return dn;

}

public static DataNode instantiateDataNode(String args [], Configuration conf, SecureResources resources) throws IOException {

return makeInstance(dataLocations, conf, resources);

}

static DataNode makeInstance(Collection dataDirs,

Configuration conf, SecureResources resources) throws IOException {

return new DataNode(conf, locations, resources); 29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

}

DataNode(final Configuration conf,

final List dataDirs,

final SecureResources resources) throws IOException {

startDataNode(conf, dataDirs, resources);

}

public void runDatanodeDaemon() throws IOException {

ll();

// start dataXceiveServer

();

if (localDataXceiverServer != null) {

();

}

();

startPlugins(conf);

}

}

上面的代碼跟蹤不難，但是最終需要注意兩個(gè)方法：startDataNode和runDatanodeDaemon方法，前面一個(gè)用于初始化DataNode，后面一個(gè)啟動(dòng)DataNode的后臺(tái)線程，這些線程是會(huì)伴隨DataNode進(jìn)程一直跑著的。接著，讓我們重點(diǎn)研究下方法startDataNode，看代碼：

?

1

2

3

4

5

6

7

8

void startDataNode(Configuration conf,

List dataDirs,

// DatanodeProtocol namenode,

SecureResources resources

) throws IOException {

storage = new DataStorage();

// global DN ttings 9

10

11

12

13

14

15

16

17

18

registerMXBean();

initDataXceiver(conf);

startInfoServer(conf);

pauMonitor = new JvmPauMonitor(conf);

();

initIpcServer(conf);

blockPoolManager = new BlockPoolManager(this);

hNamenodes(conf);

}

registerMXBean這個(gè)方法可以忽略，用來(lái)注冊(cè)MBean信息；initDataXceiver這個(gè)方法應(yīng)該來(lái)說(shuō)還是比較重要，實(shí)例化的dataXceiverServer用來(lái)接受客戶端或者其他datanode的數(shù)據(jù)接收或者發(fā)送請(qǐng)求；startInfoServer方法用來(lái)啟動(dòng)datanode的web服務(wù)器；pauMonitor用來(lái)監(jiān)控jvm是否有停頓；initIpcServer方法比較重要，用來(lái)啟動(dòng)datanode上的rpc服務(wù)，主要包括兩個(gè)服務(wù)：ClientDatanodeProtocolPB和InterDatanodeProtocolPB。

然后屬于DataNode的重點(diǎn)來(lái)了，blockPoolManager對(duì)象的實(shí)例化，注意一點(diǎn)，2.4.1 這個(gè)版本的hadoop已經(jīng)支持了hadoop Federation的特性，而blockPooolManager就是支撐這個(gè)特性來(lái)的。現(xiàn)在讓我們來(lái)看看他里面的東西。還是先上代碼吧。

?

1

2

3

4

5

6

7

8

class BlockPoolManager {

BlockPoolManager(DataNode dn) {

= dn;

}

void refreshNamenodes(Configuration conf)

throws IOException {

synchronized (refreshNamenodesLock) { 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

doRefreshNamenodes(newAddressMap);

}

}

private void doRefreshNamenodes(

Map> addrMap) throws IOException {

synchronized (this) {

startAll();

}

}

synchronized void startAll() throws IOException {

for (BPOfferService bpos : offerServices) {

();

}

}

}

class BPOfferService {

void start() {

for (BPServiceActor actor : bpServices) {

();

}

}

}

class BPServiceActor implements Runnable {

void start() {

if ((bpThread != null) && (e())) {

//Thread is started already

return;

}

bpThread = new Thread(this, formatThreadName());

mon(true); // needed for JUnit testing

();

}

public void run() {

while (true) {

connectToNNAndHandshake();

break;

}

while (shouldRun()) { 53

54

55

56

57

58

59

60

61

62

63

64

offerService();

}

}

private void offerService() throws Exception {

while (shouldRun()) {

HeartbeatRespon resp = ndHeartBeat();

List cmds = blockReport();

}

}

}

順著代碼往下走，整個(gè)思路都會(huì)比較清晰了，BPServiceActor這個(gè)類做了具體的事情，包括datanode跟namenode的握手，發(fā)送心跳和報(bào)告塊信息，執(zhí)行namenode發(fā)回來(lái)的命名。

詳細(xì)的過(guò)程就不啰嗦了。

到這里DataNode的啟動(dòng)過(guò)程就搞了一個(gè)段落。

-