日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Hadoop源碼分析之心跳機制

一．心跳機制

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)鍵代碼：

代碼一：

 
 
 
    
  
  
  
  1. /**  
    2. 
  
  
  
  2.    * 循環(huán)調(diào)用“發(fā)送心跳”方法，直到shutdown  
    3. 
  
  
  
  3.    * 調(diào)用遠程Namenode的方法  
    4. 
  
  
  
  4.    */ 
    5. 
  
  
  
  5.   public void offerService() throws Exception {  
    6. 
  
  
  
  6. ???  
    7. 
  
  
  
  7.     while (shouldRun) {  
    8. 
  
  
  
  8.       try {  
    9. 
  
  
  
  9.         long startTime = now();  
    10. 
  
  
  
  10.          // heartBeatInterval是在啟動Datanode時根據(jù)配置文件設(shè)置的，是心跳間隔時間  
    11. 
  
  
  
  11.         if (startTime - lastHeartbeat > heartBeatInterval) {  
    12. 
  
  
  
  12.           lastHeartbeat = startTime;  
    13. 
  
  
  
  13. //Datanode發(fā)送心跳  
    14. 
  
  
  
  14.           DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,  
    15. 
  
  
  
  15.                                                        data.getCapacity(),  
    16. 
  
  
  
  16.                                                        data.getDfsUsed(),  
    17. 
  
  
  
  17.                                                        data.getRemaining(),  
    18. 
  
  
  
  18.                                                        xmitsInProgress.get(),  
    19. 
  
  
  
  19.                                                        getXceiverCount());  
    20. 
  
  
  
  20.           myMetrics.addHeartBeat(now() - startTime);  
    21. 
  
  
  
  21.            
    22. 
  
  
  
  22.           if (!processCommand(cmds))  
    23. 
  
  
  
  23.             continue;  
    24. 
  
  
  
  24.         }  
    25. 
  
  
  
  25.          
    26. 
  
  
  
  26.       ???  
    27. 
  
  
  
  27.       }  
    28. 
  
  
  
  28.     } // while (shouldRun)  
    29. 
  
  
  
  29.   } // offerService 
    30.

需要注意的是：發(fā)送心跳的對象并不是datanode，而是一個名為namenode的對象，難道在datanode端就直接有個namenode的引用嗎？其實不然，我們來看看這個namenode吧：

代碼二：

 
 
 
    
  
  
  
  1. public DatanodeProtocol namenode = null; 
    2.

namenode其實是一個DatanodeProtocol的引用，在對hadoop RPC機制分析的文章中我提到過，這是一個Datanode和Namenode通信的協(xié)議，其中有許多未實現(xiàn)的接口方法，sendHeartbeat()就是其中的一個。下面看看這個namenode對象是怎么被實例化的吧：

代碼三：

 
 
 
    
  
  
  
  1. this.namenode = (DatanodeProtocol)   
    2. 
  
  
  
  2.     RPC.waitForProxy(DatanodeProtocol.class,  
    3. 
  
  
  
  3.                      DatanodeProtocol.versionID,  
    4. 
  
  
  
  4.                      nameNodeAddr,   
    5. 
  
  
  
  5.                      conf); 
    6.

其實這個namenode并不是Namenode的一個對象，而只是一個Datanode端對Namenode的代理對象，正是這個代理完成了“心跳”。代理的底層實現(xiàn)就是RPC機制了。

三．Tasktracker、Jobtracker心跳源碼分析

同樣我們從Tasktracker入手，下面貼出Tasktracker.class的關(guān)鍵代碼：

代碼四：

 
 
 
    
  
  
  
  1. 代碼一：  
    2. 
  
  
  
  2. State offerService() throws Exception {  
    3. 
  
  
  
  3.     long lastHeartbeat = System.currentTimeMillis();  
    4. 
  
  
  
  4.     while (running && !shuttingDown) {  
    5. 
  
  
  
  5.      ???  
    6. 
  
  
  
  6.           
    7. 
  
  
  
  7.         // 發(fā)送心跳，調(diào)用代碼二  
    8. 
  
  
  
  8.         HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);  
    9. 
  
  
  
  9.  
    10. 
  
  
  
  10.       ???  
    11. 
  
  
  
  11.     return State.NORMAL;  
    12. 
  
  
  
  12.   }  
    13. 
  
  
  
  13.  
    14. 
  
  
  
  14. 代碼二：  
    15. 
  
  
  
  15. HeartbeatResponse transmitHeartBeat(long now) throws IOException {  
    16. 
  
  
  
  16.    ???  
    17. 
  
  
  
  17.     HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,   
    18. 
  
  
  
  18.                                                               justStarted,  
    19. 
  
  
  
  19.                                                               justInited,  
    20. 
  
  
  
  20.                                                               askForNewTask,   
    21. 
  
  
  
  21.                                                          heartbeatResponseId);                                
    22. 
  
  
  
  22. ???  
    23. 
  
  
  
  23.     return heartbeatResponse;  
    24. 
  
  
  
  24.   } 
    25.

其實我覺得分析到這里大家就可以自己分析了，jobClient也是一個協(xié)議：

代碼五：

 
 
 
    
  
  
  
  1. InterTrackerProtocol jobClient; 
    2.

該協(xié)議用于定義Tasktracker和Jobtracker的通信。同樣，它也是一個代理對象：

代碼六：

 
 
 
    
  
  
  
  1. this.jobClient = (InterTrackerProtocol)   
    2. 
  
  
  
  2.     UserGroupInformation.getLoginUser().doAs(  
    3. 
  
  
  
  3.         new PrivilegedExceptionAction() {  
  
  
  
  4.       public Object run() throws IOException {  
    5. 
  
  
  
  5.         return RPC.waitForProxy(InterTrackerProtocol.class,  
    6. 
  
  
  
  6.             InterTrackerProtocol.versionID,  
    7. 
  
  
  
  7.             jobTrackAddr, fConf);  
    8. 
  
  
  
  8.       }  
    9. 
  
  
  
  9.     }); 
    10. 
 
 
 
    終于，hadoop底層通信整個系列的源碼分析全部完成了。我可以好好地復(fù)習(xí)學(xué)校的功課了。呵呵。
    原文鏈接：http://www.blogjava.net/jlins-you/archive/2012/05/31/379603.html
                                                    
    
                                                網(wǎng)頁標(biāo)題：Hadoop源碼分析之心跳機制                                                
    
                                                文章URL：http://www.5511xx.com/article/cojjsss.html