一．心跳機制
1. hadoop集群是master/slave模式，master包括Namenode和Jobtracker，slave包括Datanode和Tasktracker。
2. master啟動的時候，會開一個ipc server在那里，等待slave心跳。
3. slave啟動時，會連接master，并每隔3秒鐘主動向master發(fā)送一個“心跳”，這個時間可 以通過”heartbeat.recheck.interval”屬性來設(shè)置。將自己的狀態(tài)信息告訴master，然后master也是通過這個心跳的返回值，向slave節(jié)點傳達指令。
4. 需要指出的是：namenode與datanode之間的通信，jobtracker與tasktracker之間的通信，都是通過“心跳”完成的。
二．Datanode、Namenode心跳源碼分析
既然“心跳”是Datanode主動給Namenode發(fā)送的。那Datanode是怎么樣發(fā)送的呢？下面貼出Datanode.class中的關(guān)鍵代碼：
代碼一：
/** * 循環(huán)調(diào)用“發(fā)送心跳”方法，直到shutdown * 調(diào)用遠程Namenode的方法 */ public void offerService() throws Exception { ••• while (shouldRun) { try { long startTime = now(); // heartBeatInterval是在啟動Datanode時根據(jù)配置文件設(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ā)送心跳的對象并不是datanode，而是一個名為namenode的對象，難道在datanode端就直接有個namenode的引用嗎？其實不然，我們來看看這個namenode吧：
代碼二：
public DatanodeProtocol namenode = null;
namenode其實是一個DatanodeProtocol的引用，在對hadoop RPC機制分析的文章中我提到過，這是一個Datanode和Namenode通信的協(xié)議，其中有許多未實現(xiàn)的接口方法，sendHeartbeat()就是其中的一個。下面看看這個namenode對象是怎么被實例化的吧：
代碼三：
this.namenode = (DatanodeProtocol) RPC.waitForProxy(DatanodeProtocol.class, DatanodeProtocol.versionID, nameNodeAddr, conf);
其實這個namenode并不是Namenode的一個對象，而只是一個Datanode端對Namenode的代理對象，正是這個代理完成了“心跳”。代理的底層實現(xiàn)就是RPC機制了。
三．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; }
其實我覺得分析到這里大家就可以自己分析了，jobClient也是一個協(xié)議：
代碼五：
InterTrackerProtocol jobClient;
該協(xié)議用于定義Tasktracker和Jobtracker的通信。同樣，它也是一個代理對象：
代碼六：
this.jobClient = (InterTrackerProtocol) UserGroupInformation.getLoginUser().doAs( new PrivilegedExceptionAction<Object>() { public Object run() throws IOException { return RPC.waitForProxy(InterTrackerProtocol.class, InterTrackerProtocol.versionID, jobTrackAddr, fConf); } });
終于，hadoop底層通信整個系列的源碼分析全部完成了。我可以好好地復習學校的功課了。呵呵。