DataNode启动流程分析

从datanode的角色来看，datanode负责是存储数据的服务，所以datanode启动时必须启动如下服务：

1，心跳服务（负责将自身状况报告给namenode。必须有，否则无法知道服务是否正常）

2，块汇报服务。（将自身存储的块信息汇报给namenode）（其中块汇报又分为增量汇报和全量汇报）

3，块检测服务。（检测磁盘是否损坏）

4，初始化存储服务。

5，接收读写数据的服务。（不用多说，datanode必备的功能）

源码追踪如下：

• DataNode.main()→
• DataNode.createDataNode()→
• DataNode.startDataNode()→
• DataNode.createDataNode()→
• DataNode.runDatanodeDaemon()→
  • BPOfferService.start()→
  • DataXceiverServer.start()

通过在namenode端增加callqueue的打印日志

可发现datanode在启动时会和namenode进行三次通信，，分别是：

1，versionRuquest。

2，registerDatanode。

3，sendHeartbeat.

现在分别对这三次rpc调用展开说：

1，versionRequest

用nn返回的信息和web页面核对，信息一致

checkNNVersion() 方法中，对nn返回的版本和dn的版本进行比对，版本不一致即打印出相关信息。

从以上可以看出 versionRequest()的作用是 获取namenode的版本信息，从而可以在页面或日志就知晓namenode和datanode的版本信息。

2，registerDatanode

通过以上代码可以看到，registerDatanode的作用如其名：是dn将自己注册到nn上，目的是为了获取nn的准入。

3，sendHeartbeat

通过代码可以看到 dn在对namenode发起sendHeartBeat后，namenode返回HeartbeatResponse对象给datanode，该对象中有需要执行的对象和 fullBlockReportLeaseId 等。 这步的作用就是借用dn发送心跳的请求将需要执行的命令返回给dn。

心跳服务是hdfs中比较重要的服务，现对该服务的流程展开详细说明：

1. 按配置，发送心跳给nanmenode
2. namenode返回 HeartbeatResponse对象给datanode
3. datanode拿到 HeartbeatResponse对象中的DatanodeCommand并执行

相关代码如下：

BPServiceActor类

HeartbeatResponse sendHeartBeat(boolean requestBlockReportLease) throws IOException { scheduler.scheduleNextHeartbeat(); StorageReport[] reports = dn.getFSDataset().getStorageReports(bpos.getBlockPoolId()); if (LOG.isDebugEnabled()) { LOG.debug("Sending heartbeat with " + reports.length + " storage reports from service actor: " + this); } VolumeFailureSummary volumeFailureSummary = dn.getFSDataset() .getVolumeFailureSummary(); int numFailedVolumes = volumeFailureSummary != null ? volumeFailureSummary.getFailedStorageLocations().length : 0; return bpNamenode.sendHeartbeat(bpRegistration, reports, dn.getFSDataset().getCacheCapacity(), dn.getFSDataset().getCacheUsed(), dn.getXmitsInProgress(), dn.getXceiverCount(), numFailedVolumes, volumeFailureSummary, requestBlockReportLease); }

public class HeartbeatResponse { /** Commands returned from the namenode to the datanode */ private final DatanodeCommand[] commands; /** Information about the current HA-related state of the NN */ private final NNHAStatusHeartbeat haStatus; private final RollingUpgradeStatus rollingUpdateStatus; private final long fullBlockReportLeaseId; public HeartbeatResponse(DatanodeCommand[] cmds, NNHAStatusHeartbeat haStatus, RollingUpgradeStatus rollingUpdateStatus, long fullBlockReportLeaseId) { commands = cmds; this.haStatus = haStatus; this.rollingUpdateStatus = rollingUpdateStatus; this.fullBlockReportLeaseId = fullBlockReportLeaseId; } public DatanodeCommand[] getCommands() { return commands; } public NNHAStatusHeartbeat getNameNodeHaState() { return haStatus; } public RollingUpgradeStatus getRollingUpdateStatus() { return rollingUpdateStatus; } public long getFullBlockReportLeaseId() { return fullBlockReportLeaseId; } }

从上面的代码可以看出，datanode在发送心跳给namenode的时候，namenode会返回执行命令，dn会执行这些命令。

2，写入数据的服务（待补充）

3，块汇报

块汇报又分增量汇报和全量汇报

全量汇报：1，namenode成为active后，会要求datanode汇报一次block信息

增量汇报（待补充）

块检测服务：

块检测服务每隔几个小时就会检测磁盘是否损坏

public class BlockScanner { public static final Logger LOG = LoggerFactory.getLogger(BlockScanner.class); /** * The DataNode that this scanner is associated with. */ private final DataNode datanode; /** * Maps Storage IDs to VolumeScanner objects. */ private final TreeMap<String, VolumeScanner> scanners = new TreeMap<String, VolumeScanner>(); /** * The scanner configuration. */ private Conf conf; }

/** * The cached scanner configuration. */ static class Conf { // These are a few internal configuration keys used for unit tests. // They can't be set unless the static boolean allowUnitTestSettings has // been set to true. @VisibleForTesting static final String INTERNAL_DFS_DATANODE_SCAN_PERIOD_MS = "in ternal.dfs.datanode.scan.period.ms.key"; @VisibleForTesting static final String INTERNAL_VOLUME_SCANNER_SCAN_RESULT_HANDLER = "internal.volume.scanner.scan.result.handler"; @VisibleForTesting static final String INTERNAL_DFS_BLOCK_SCANNER_MAX_STALENESS_MS = "internal.dfs.block.scanner.max_staleness.ms"; ...... }