資訊專欄INFORMATION COLUMN
摘要：博客原文最近集群遇到一個問題就是集群在跑任務的時候，會超時而被，但任務重跑則成功，問題是隨機的出現(xiàn)的，所以初步懷疑是因為心跳匯報出現(xiàn)問題或則因為繁忙住，因為某些機制導致等待不匯報心跳，所以我們還是先了解，是如何向匯報心跳的。
博客原文
hackshell
最近集群遇到一個問題,就是集群在跑任務的時候，AM會超時10min而被KILL，但任務重跑則成功，問題是隨機的出現(xiàn)的，所以初步懷疑是因為AM心跳匯報出現(xiàn)問題或則RM因為繁忙hang住，AM因為某些機制導致等待10min不匯報心跳，所以我們還是先了解，AM是如何向RM匯報心跳的。
在MRAppMaster中，ContainerAllocatorRouter負責向RM申請資源（發(fā)送心跳）
RMContainerAllocator其最終父類是RMCommunicator，它實現(xiàn)了RMHeartbeatHandler接口
public interface RMHeartbeatHandler {
long getLastHeartbeatTime(); // 獲取上一次心跳的時間
void runOnNextHeartbeat(Runnable callback); // 回調(diào)注冊到callback隊列的callback函數(shù)
}
每一次心跳回來，都會執(zhí)行一次注冊在heartbeatCallbacks中的回調(diào)函數(shù)：
allocatorThread = new Thread(new Runnable() {
@Override
public void run() {
while (!stopped.get() && !Thread.currentThread().isInterrupted()) {
......
heartbeat();
lastHeartbeatTime = context.getClock().getTime();// 記錄上一次心跳時間
executeHeartbeatCallbacks(); // 執(zhí)行回調(diào)函數(shù)
....
});
RMCommunicator類中：
private void executeHeartbeatCallbacks() {
Runnable callback = null;
while ((callback = heartbeatCallbacks.poll()) != null) {
callback.run();
}
}
在RMCommunicator啟動時，首先會向RM注冊，把自己的host和port告訴RM，然后在啟動一條線程（startAllocatorThread）定期的調(diào)用RMContainerAllocator中實現(xiàn)的heartbeat方法（向RM申請資源，定期匯報信息，告訴RM自己還活著）。
AM初始化同時也會初始化RMCommunicator：
protected void serviceStart() throws Exception {
scheduler= createSchedulerProxy(); // 獲取RM的代理
register(); // 注冊
startAllocatorThread(); // 心跳線程
....
}
AM的ContainerAllocatorRouter事件處理流程如下圖：
注冊流程：
調(diào)用RMCommunicator遠程調(diào)用ApplicationMasterService的registerApplicationMaster方法，設置維護responseId，然后把它加入AMLivelinessMonitor中，并使用map記錄時間，用來監(jiān)控AM是否因為長時間沒有心跳而超時，如果AM長時間沒有心跳信息更新，RM就會通知NodeManager把AM移除。
心跳線程：
在發(fā)送心跳的過程中，即也是獲取資源的過程
@Override
protected synchronized void heartbeat() throws Exception {
scheduleStats.updateAndLogIfChanged("Before Scheduling: ");
List allocatedContainers = getResources();// 重要的方法
if (allocatedContainers.size() > 0) {
scheduledRequests.assign(allocatedContainers);
}
......
}
獲取資源的過程：
private ListgetResources() throws Exception { ... response = makeRemoteRequest(); // 和RM進行交互 ... // 優(yōu)先處理RM發(fā)送過來的命令 if (response.getAMCommand() != null) { switch(response.getAMCommand()) { case AM_RESYNC: case AM_SHUTDOWN: eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.JOB_AM_REBOOT)); throw new YarnRuntimeException("Resource Manager doesn"t recognize AttemptId: " + this.getContext().getApplicationID()); default: .... } // 等等一系列處理 } }
構(gòu)建請求：
protected AllocateResponse makeRemoteRequest() throws IOException {
AllocateRequest allocateRequest =
AllocateRequest.newInstance(lastResponseID,
super.getApplicationProgress(), new ArrayList(ask),
new ArrayList(release), blacklistRequest);
AllocateResponse allocateResponse;
allocateResponse = scheduler.allocate(allocateRequest); // RPC調(diào)用ApplicationMasterService的allocate方法
.....
}
每一次心跳的調(diào)用都會刷新AMLivelinessMonitor的時間，代表AM還活著
而且我們通過代碼可以看出，資源請求被封裝為一個ask，即一個ResourceRequest的ArrayList的資源列表 例如：
