Hadoop源代码分析（一七）

xiu12

周围的障碍扫清以后，我们可以开始分析类DataNode。类图如下：

public class DataNode extends Configured
    implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable

　　上面给出了DataNode的继承关系，我们发现，DataNode实现了两个通信接口，其中ClientDatanodeProtocol是用于和Client交互的，InterDatanodeProtocol，就是我们前面提到的DataNode间的通信接口。ipcServer（类图的左下方）是DataNode的一个成员变量，它启动了一个IPC服务，这样，DataNode就能提供ClientDatanodeProtocol和InterDatanodeProtocol的能力了。
　　我们从main函数开始吧。这个函数很简单，调用了createDataNode的方法，然后就等着DataNode的线程结束。createDataNode首先调用instantiateDataNode初始化DataNode，然后执行runDatanodeDaemon。runDatanodeDaemon会向NameNode注册，如果成功，才启动DataNode线程，DataNode就开始干活了。
　　初始化DataNode的方法instantiateDataNode会读取DataNode需要的配置文件，同时读取配置的storage目录（可能有多个，看storage的讨论部分），然后把这两参数送到makeInstance中，makeInstance会先检查目录（存在，是目录，可读，可写），然后调用：
new DataNode(conf, dirs);
　　接下来控制流就到了构造函数上。构造函数调用startDataNode，完成和DataNode相关的初始化工作（注意，DataNode工作线程不在这个函数里启动）。首先是初始化一堆的配置参数，什么NameNode地址，socket参数等等。然后，向NameNode请求配置信息（DatanodeProtocol.versionRequest），并检查返回的NamespaceInfo和本地的版本是否一致。
　　正常情况的下一步是检查文件系统的状态并做必要的恢复，初始化FSDataset（到这个时候，上面图中storage和data成员变量已经初始化）。
　　然后，找一个端口并创建DataXceiverServer（run方法里启动），创建DataBlockScanner（根据需要在offerService中启动，只启动一次），创建DataNode上的HttpServer，启动ipcServer。这样就结束了DataNode相关的初始化工作。
　　在启动DataNode工作线程前，DataNode需要向NameNode注册。注册信息在初始化的时候已经构造完毕，包括DataXceiverServer端口，ipcServer端口，文件布局版本号等重要信息。注册成功后就可以启动DataNode线程。
　　DataNode的run方法，循环里有两种选择，升级（暂时不讨论）/正常工作。我们来看正常工作的offerService方法。offerService也是个循环，在循环里，offerService会定时向NameNode发送心跳，报告系统中Block状态的变化，报告DataNode现在管理的Block状态。发送心跳和Block状态报告时，NameNode会返回一些命令，DataNode将执行这些命令。
　　心跳的处理比较简单，以heartBeatInterval间隔发送。
　　Block状态变化报告，会利用保存在receivedBlockList和delHints两个列表中的信息。receivedBlockList表明在这个DataNode成功创建的新的数据块，而delHints，是可以删除该数据块的节点。如在DataXceiver的replaceBlock中，有调用：
datanode.notifyNamenodeReceivedBlock(block, sourceID)
　　这表明，DataNode已经从sourceID上接收了一个Block，sourceID上对应的Block可以删除了（这个场景出现在当系统需要做负载均衡时，Block在DataNode之间拷贝）。
　　Block状态变化报告通过NameNode.blockReceived来报告。
　　Block状态报告也比较简单，以blockReportInterval间隔发送。
　　心跳和Block状态报告可以返回命令，这也是NameNode先DataNode发起请求的唯一方法。我们来看一下都有那些命令：

  DNA_TRANSFER：拷贝数据块到其他DataNode

  DNA_INVALIDATE：删除数据块（简单方法）

  DNA_SHUTDOWN：关闭DataNode（简单方法）

  DNA_REGISTER：DataNode重新注册（简单方法）

  DNA_FINALIZE：提交升级（简单方法）

  DNA_RECOVERBLOCK：恢复数据块

　　拷贝数据块到其他DataNode由transferBlocks方法执行。注意，返回的命令可以包含多个数据块，每一个数据块可以包含多个目标地址。transferBlocks方法将为每一个Block启动一个DataTransfer线程，用于传输数据。
　　DataTransfer是一个DataNode的内部类，它利用我们前面介绍的OP_WRITE_BLOCK写数据块操作，发送数据到多个目标上面。
　　恢复数据块和NameNode的租约（lease）恢复有关，我们后面再讨论。