摘要：博客原文最近集群遇到一个问题就是集群在跑任务的时候，会超时而被，但任务重跑则成功，问题是随机的出现的，所以初步怀疑是因为心跳汇报出现问题或则因为繁忙住，因为某些机制导致等待不汇报心跳，所以我们还是先了解，是如何向汇报心跳的。
博客原文
hackshell
最近集群遇到一个问题,就是集群在跑任务的时候，AM会超时10min而被KILL，但任务重跑则成功，问题是随机的出现的，所以初步怀疑是因为AM心跳汇报出现问题或则RM因为繁忙hang住，AM因为某些机制导致等待10min不汇报心跳，所以我们还是先了解，AM是如何向RM汇报心跳的。
在MRAppMaster中，ContainerAllocatorRouter负责向RM申请资源（发送心跳）
RMContainerAllocator其最终父类是RMCommunicator，它实现了RMHeartbeatHandler接口
public interface RMHeartbeatHandler {
long getLastHeartbeatTime(); // 获取上一次心跳的时间
void runOnNextHeartbeat(Runnable callback); // 回调注册到callback队列的callback函数
}
每一次心跳回来，都会执行一次注册在heartbeatCallbacks中的回调函数：
allocatorThread = new Thread(new Runnable() {
@Override
public void run() {
while (!stopped.get() && !Thread.currentThread().isInterrupted()) {
......
heartbeat();
lastHeartbeatTime = context.getClock().getTime();// 记录上一次心跳时间
executeHeartbeatCallbacks(); // 执行回调函数
....
});
RMCommunicator类中：
private void executeHeartbeatCallbacks() {
Runnable callback = null;
while ((callback = heartbeatCallbacks.poll()) != null) {
callback.run();
}
}
在RMCommunicator启动时，首先会向RM注册，把自己的host和port告诉RM，然后在启动一条线程（startAllocatorThread）定期的调用RMContainerAllocator中实现的heartbeat方法（向RM申请资源，定期汇报信息，告诉RM自己还活着）。
AM初始化同时也会初始化RMCommunicator：
protected void serviceStart() throws Exception {
scheduler= createSchedulerProxy(); // 获取RM的代理
register(); // 注册
startAllocatorThread(); // 心跳线程
....
}
AM的ContainerAllocatorRouter事件处理流程如下图：
注册流程：
调用RMCommunicator远程调用ApplicationMasterService的registerApplicationMaster方法，设置维护responseId，然后把它加入AMLivelinessMonitor中，并使用map记录时间，用来监控AM是否因为长时间没有心跳而超时，如果AM长时间没有心跳信息更新，RM就会通知NodeManager把AM移除。
心跳线程：
在发送心跳的过程中，即也是获取资源的过程
@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进行交互 ... // 优先处理RM发送过来的命令 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: .... } // 等等一系列处理 } }
构建请求：
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调用ApplicationMasterService的allocate方法
.....
}
每一次心跳的调用都会刷新AMLivelinessMonitor的时间，代表AM还活着
而且我们通过代码可以看出，资源请求被封装为一个ask，即一个ResourceRequest的ArrayList的资源列表 例如：
