這篇文章將為大家詳細(xì)講解有關(guān)怎么用Hadoop源碼分析心跳機(jī)制，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

創(chuàng)新互聯(lián)公司2013年至今，公司以做網(wǎng)站、成都網(wǎng)站設(shè)計(jì)、系統(tǒng)開(kāi)發(fā)、網(wǎng)絡(luò)推廣、文化傳媒、企業(yè)宣傳、平面廣告設(shè)計(jì)等為主要業(yè)務(wù)，適用行業(yè)近百種。服務(wù)企業(yè)客戶近千家，涉及國(guó)內(nèi)多個(gè)省份客戶。擁有多年網(wǎng)站建設(shè)開(kāi)發(fā)經(jīng)驗(yàn)。為企業(yè)提供專業(yè)的網(wǎng)站建設(shè)、創(chuàng)意設(shè)計(jì)、宣傳推廣等服務(wù)。 通過(guò)專業(yè)的設(shè)計(jì)、獨(dú)特的風(fēng)格，為不同客戶提供各種風(fēng)格的特色服務(wù)。

一．心跳機(jī)制

1. hadoop集群是master/slave模式，master包括Namenode和Jobtracker，slave包括Datanode和Tasktracker。

2. master啟動(dòng)的時(shí)候，會(huì)開(kāi)一個(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