Hadoop源碼分析之心跳機(jī)制
一.心跳機(jī)制
1. hadoop集群是master/slave模式,master包括Namenode和Jobtracker,slave包括Datanode和Tasktracker。
2. master啟動(dòng)的時(shí)候,會(huì)開一個(gè)ipc server在那里,等待slave心跳。
3. slave啟動(dòng)時(shí),會(huì)連接master,并每隔3秒鐘主動(dòng)向master發(fā)送一個(gè)“心跳”,這個(gè)時(shí)間可 以通過(guò)”heartbeat.recheck.interval”屬性來(lái)設(shè)置。將自己的狀態(tài)信息告訴master,然后master也是通過(guò)這個(gè)心跳的返回值,向slave節(jié)點(diǎn)傳達(dá)指令。
4. 需要指出的是:namenode與datanode之間的通信,jobtracker與tasktracker之間的通信,都是通過(guò)“心跳”完成的。
二.Datanode、Namenode心跳源碼分析
既然“心跳”是Datanode主動(dòng)給Namenode發(fā)送的。那Datanode是怎么樣發(fā)送的呢?下面貼出Datanode.class中的關(guān)鍵代碼:
代碼一:
- /**
- * 循環(huán)調(diào)用“發(fā)送心跳”方法,直到shutdown
- * 調(diào)用遠(yuǎn)程N(yùn)amenode的方法
- */
- public void offerService() throws Exception {
- •••
- while (shouldRun) {
- try {
- long startTime = now();
- // heartBeatInterval是在啟動(dòng)Datanode時(shí)根據(jù)配置文件設(shè)置的,是心跳間隔時(shí)間
- if (startTime - lastHeartbeat > heartBeatInterval) {
- lastHeartbeat = startTime;
- //Datanode發(fā)送心跳
- DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
- data.getCapacity(),
- data.getDfsUsed(),
- data.getRemaining(),
- xmitsInProgress.get(),
- getXceiverCount());
- myMetrics.addHeartBeat(now() - startTime);
- if (!processCommand(cmds))
- continue;
- }
- •••
- }
- } // while (shouldRun)
- } // offerService
需要注意的是:發(fā)送心跳的對(duì)象并不是datanode,而是一個(gè)名為namenode的對(duì)象,難道在datanode端就直接有個(gè)namenode的引用嗎?其實(shí)不然,我們來(lái)看看這個(gè)namenode吧:
代碼二:
- public DatanodeProtocol namenode = null;
namenode其實(shí)是一個(gè)DatanodeProtocol的引用,在對(duì)hadoop RPC機(jī)制分析的文章中我提到過(guò),這是一個(gè)Datanode和Namenode通信的協(xié)議,其中有許多未實(shí)現(xiàn)的接口方法,sendHeartbeat()就是其中的一個(gè)。下面看看這個(gè)namenode對(duì)象是怎么被實(shí)例化的吧:
代碼三:
- this.namenode = (DatanodeProtocol)
- RPC.waitForProxy(DatanodeProtocol.class,
- DatanodeProtocol.versionID,
- nameNodeAddr,
- conf);
其實(shí)這個(gè)namenode并不是Namenode的一個(gè)對(duì)象,而只是一個(gè)Datanode端對(duì)Namenode的代理對(duì)象,正是這個(gè)代理完成了“心跳”。代理的底層實(shí)現(xiàn)就是RPC機(jī)制了。
三.Tasktracker、Jobtracker心跳源碼分析
同樣我們從Tasktracker入手,下面貼出Tasktracker.class的關(guān)鍵代碼:
代碼四:
- 代碼一:
- State offerService() throws Exception {
- long lastHeartbeat = System.currentTimeMillis();
- while (running && !shuttingDown) {
- •••
- // 發(fā)送心跳,調(diào)用代碼二
- HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);
- •••
- return State.NORMAL;
- }
- 代碼二:
- HeartbeatResponse transmitHeartBeat(long now) throws IOException {
- •••
- HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,
- justStarted,
- justInited,
- askForNewTask,
- heartbeatResponseId);
- •••
- return heartbeatResponse;
- }
其實(shí)我覺(jué)得分析到這里大家就可以自己分析了,jobClient也是一個(gè)協(xié)議:
代碼五:
- InterTrackerProtocol jobClient;
該協(xié)議用于定義Tasktracker和Jobtracker的通信。同樣,它也是一個(gè)代理對(duì)象:
代碼六:
- this.jobClient = (InterTrackerProtocol)
- UserGroupInformation.getLoginUser().doAs(
- new PrivilegedExceptionAction<Object>() {
- public Object run() throws IOException {
- return RPC.waitForProxy(InterTrackerProtocol.class,
- InterTrackerProtocol.versionID,
- jobTrackAddr, fConf);
- }
- });
終于,hadoop底層通信整個(gè)系列的源碼分析全部完成了。我可以好好地復(fù)習(xí)學(xué)校的功課了。呵呵。
原文鏈接:http://www.blogjava.net/jlins-you/archive/2012/05/31/379603.html