Hadoop 中的 NameNode 和 DataNode 之间的心跳机制

arongsoft

　　最近需要在 Hive 中实现 C/S 端的心跳，而心跳在 Hadoop 中被广泛使用，因此首先要清楚 Hadoop 中的心跳是如何实现的。在 Hadoop 中，NameNode 和 DataNode 之间、JobTracker 和 TaskTracker 之间都会有心跳信息，选取前者进行分析。


Hadoop 中通过心跳消息来确定 DataNodes 的有效性。对于超过时间未收到心跳消息的 DataNodes，系统认定这些 DataNodes 已经失效，将其从系统中移除。


心跳消息由 DataNode 发起，DataNode 是一个 Runnable 的对象，在系统运行的时候作为 Daemon 运行，


public static void runDatanodeDaemon(DataNode dn) throws IOException {
if (dn != null) {
//register datanode
dn.register();
dn.dataNodeThread = new Thread(dn, dnThreadName);
dn.dataNodeThread.setDaemon(true); // needed for JUnit testing
dn.dataNodeThread.start();
}
}
　　

在 DataNode 的 run() 方法中会调用 offerservice() 方法，而在这个方法中会

定期发送心跳消息：

  public void offerService() throws Exception {
...
while (shouldRun) {
...
if (startTime - lastHeartbeat > heartBeatInterval) {
//
// All heartbeat messages include following info:
// -- Datanode name
// -- data transfer port
// -- Total capacity
// -- Bytes remaining
//
lastHeartbeat = startTime;
DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
data.getCapacity(),
data.getDfsUsed(),
data.getRemaining(),
xmitsInProgress.get(),
getXceiverCount());
myMetrics.heartbeats.inc(now() - startTime);
//LOG.info("Just sent heartbeat, with name " + localName);
if (!processCommand(cmds))
continue;
}
...
}
}
 　　上述代码中的 namenode.sendHeartbeat() 方法是一个 对 namenode 的 RPC 调用，在 NameNode 类中的 sendHeartbeat() 方法中被处理。 该方法将最终调用 namesystem.handleHeartbeat(...) 来对心跳消息进行处理，更新心跳消息列表。


在 FSNamesystem 类中有一个 HeartbeatMonitor 子类，这是一个 Runnable 对象，用于定期的检查已经接受到的心跳消息，确定哪些 DataNode 已经失效，然后将失效的 DataNode 从系统中移除。
　　至此，一个完整的心跳消息的发起，汇总，处理的过程完成。
　　

总结：Hadoop 中的心跳消息由 DataNode 发起，在 NameNode 中进行处理，最终由 HeartbeatMonitor 进行维护。