Hadoop源代码的边角料：HDFS的数据通信机制

kient88

　　这阵子花了点时间读了读HDFS的源代码。有所得。
不过网上关于Hadoop源代码的解析已经挺多了，所以自称为“边角料”，也就是一些零散的心得和想法。
　　
　　简单而言，HDFS分为了三个部分：
NameNode，处于master的地位，维护了数据在DataNode上的分布情况，并且，还要负责一些调度任务；
DataNode，存储真实数据的地方；
DFSClient，一个client端，通过它提供的接口访问NameNode和DataNode；
三者之间的通信全部是基于TCP/Socket; 如图所示：

　　
　　图中，连线表示两者之间存在通信，箭头一方表示接收请求，没有箭头的一端表示发起请求的一方；图中的黑色细线表示控制消息的通路，红色粗线表示数据消息的通路；

可以看得出来，NameNode是一个典型的Server端程序，它总是处于接受请求，返回响应的状态中。NameNode永远不会主动的向其它组件发起请求（依稀记得GFS的论文中也是这样做的）。如果NameNode需要向DataNode发送一些调度或者控制命令的话，必须等待DataNode向NameNode发送heartbeat之后，作为heartbeat的response返回给DataNode。
DataNode就比较忙了，它不仅需要定时的发送heartbeat给NameNode，并且heartbeat的返回往往还附带了很多的控制消息需要处理，同时，DataNode要接收DFSClient来的读写数据的请求和一些控制请求，最后，DataNode之间还有数据消息和控制消息的传输。
　　
　　HDFS中很有意思的一点是，它的控制消息的传输和数据消息的传输采用的是不一样的模块。这也是接下来我想探讨的重点。
　　
　　HDFS中所有的控制消息的传输都基于它自实现的RPC模块。之前的一篇blog介绍过这个RPC的实现，这里简单回顾一下：
RPC模块会对两个需要通信的Node之间各创建一个socket，这两个Node之间的所有的控制消息都通过这对socket进行传输；对于RPC的client一端来说，会有两个线程参与到这个过程中。一个线程是调用本次RPC的线程，它会将消息写入到socket中，然后就执行wait()阻塞住；另一个线程则是RPC模块内部负责该socket读的工作，它从socket中读取消息，然后执行notify()唤醒阻塞线程；
我在之前的blog中就提到，这个机制不太适合大数据量的传输，因为两个Node之间只用一个socket进行通信，网络的吞吐量不一定上得去。
　　
　　而事实上，HDFS确实没有用RPC机制传输数据消息。当HDFS中的DFSClient对DataNode上保存的文件数据进行读写的时候，它其实采用了另外一个机制，简单介绍一下：
　　每个DataNode在启动的时候会创建一个线程DataXceiverServer来专门负责block数据的读写的链接。而DataXceiverServer做的事情很简单 --一旦有一个连接，就创建一个新的DataXceiver来处理这个连接：
　　public void run() {while (datanode.shouldRun) {try {Socket s = ss.accept();s.setTcpNoDelay(true);new Daemon(datanode.threadGroup, new DataXceiver(s, datanode, this)).start();} catch (SocketTimeoutException ignored) {// wake up to see if should continue to run} catch (IOException ie) {// ............} catch (Throwable te) {// ............}}try {ss.close();} catch (IOException ie) {// .......}}
　　DataXceiver也是一个线程，它负责处理对应的一个连接，主要完成4种任务：
opReadBlock： 读取一个block
opWriteBlock： 写一个block到disk上
opCopyBlock： 读一个block，然后送到指定的目的地
opReplaceBlock： 替换一个block
class DataXceiver extends DataTransferProtocol.Receiverimplements Runnable, FSConstants {// ................/*** Read/write data from/to the DataXceiveServer.*/public void run() {updateCurrentThreadName("Waiting for operation");DataInputStream in=null; try {in = new DataInputStream(new BufferedInputStream(NetUtils.getInputStream(s), SMALL_BUFFER_SIZE));final DataTransferProtocol.Op op = readOp(in);// Make sure the xciver count is not exceeded// ....processOp(op, in);} catch (Throwable t) {LOG.error(datanode.dnRegistration + ":DataXceiver",t);} finally {//.....}}/** Process op by the corresponding method. */protected final void processOp(Op op, DataInputStream in) throws IOException {switch(op) {case READ_BLOCK:opReadBlock(in);break;case WRITE_BLOCK:opWriteBlock(in);break;case REPLACE_BLOCK:opReplaceBlock(in);break;case COPY_BLOCK:opCopyBlock(in);break;case BLOCK_CHECKSUM:opBlockChecksum(in);break;default:throw new IOException("Unknown op " + op + " in data stream");}}
　　所以，当HDFS进行数据传输的时候，对于每一个链接创建一个thread进行处理，这样，如果两个Node之间的数据传输很频繁的话，那么有可能会创建多个链接，吞吐量就上去了。
　　
　　如果熟悉network server architecture的话，很容易知道HDFS在这里采用的是one thread per request的模型。它没有采用当下流行的基于epoll的event-driven architecture，甚至于，它都没有采用thread pool的方式，而是使用了一种“很土很土”的模型。众所周知，one thread per request模型一个很明显的缺陷在于如果访问的并发数太高，可能产生大量的thread而导致thread间的context swith开销过大。
我个人想法，HDFS之所以采用这样的一个模型，一方面是编程上比较简单，另一个方面，可能是开发人员认为HDFS这样的一个系统并不容易出现高并发的访问。从图中看，需要和DataNode进行数据消息交互的模块有两个：一个是DFSClient，一个是其它的DataNode。
先说后者，DataNode与DataNode之间的数据消息交互只发生在一种情况下，就是某个DFSClient对block进行了写操作，那么被写的DataNode需要将这些数据复制到这个block副本所在的其它DataNode上。是一种链式结构：
    DFSClient  -->DataNode A -->DataNode B --> DataNode C
所以，DataNode之间的链接是与DFSClient和DataNode之间的链接一一对应的。
再说DFSClient，它并不同于Web Server所要面对的client。Web Serve的client是终端的浏览器，可能成千上万，这是不可控的。而DFSClient是系统内的client，它的数量不会太多（就好比是对Database的连接数，是开发人员控制的，所以不会太大）。由于DFSClient在系统内部的数量不会太多，所以DataNode从DFSClient过来的连接也就不会太多。既然DFSClient发起的连接不多，那么DataNode之间的连接也不会多。
结合以上二点，整个HDFS很难产生高并发的情况，所以采用one thread per request的架构也就说得过去了。
　　
　　-- END --
　　
             版权声明：本文为博主原创文章，未经博主允许不得转载。