通过Hadoop RPC框架学习NIO
今儿个我们通过HDFS NameNode学习Java NIO框架。读本文之前假定各位大佬有NIO的基础（Selector、Channel、Buffer、以及NIO编程的模板框架）。
Hadoop RPC的Server类用到了NIO、Reactor模式等技术来尽可能的提高服务器端的并发度。可以说，学习Server类的源码是学习NIO非常好的实例，好，话不多说，让我们看看NIO是如何在HDFS这种广泛使用的大数据组件上使用的。
首先说一些前置知识：Hadoop RPC框架的服务端代码实现是在Server类。Server类下面有很多内部类。如下图所示：
Server#Listener
Listener继承了Thread类，所以它本质上也是个线程类。
Listener类主要是用来处理客户端的Socket连接，类似于Reactor模式中的Reactor角色。
Listener的成员变量如下图所示：
Listener的构造函数中进行这些成员变量的初始化，是NIO编程框架的固定写法：
接着在Listener的run方法中调用select()方法，阻塞的做select操作，只有当至少一个channel被selected，或者当前线程被interrupted，或者此selector的wakeup方法被调用时才会返回。select()最终会选择一些key的结合，这些key对应的channel是I/O就绪的。
当Listener这侧判断key是OP_ACCEPT，就知道是客户端要与服务器进行连接了，于是调用doAccept方法，并把key作为参数传入（便于后续调用attach方法分配handler）。
这里我们接着看doAccept方法，先用文字分析一下：
通过ServerSocketChannel的accept方法，拿到代表客户端Socket的SocketChannel，将客户端的SocketChannel注册到connectionManager中，ConnectionManager类用于连接的管理，包括最长空闲时间，最多连接数，Connetion对象集合等等。然后将包含客户端SocketChannel的Connection对象attach到SelectionKey上，方便后面从SelectionKey上调用attachment方法得到这个包含客户端SocketChannel的Connection对象。最后把连接添加到Reader的pendingConnections这个阻塞队列中，同时让Reader类的Selector对象readSelector立即返回，readSelector在进行下一次select之前，会把阻塞队列pendingConnections里的Connection处理完。代码如下：
void doAccept(SelectionKey key) throws InterruptedException, IOException, OutOfMemoryError {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel;
// 获取到代表客户端Socket的channel
while ((channel = server.accept()) != null) {
// NIO常用配置
channel.configureBlocking(false);
channel.socket().setTcpNoDelay(tcpNoDelay);
channel.socket().setKeepAlive(true);
// Reader相当于Reactor模式中的handler，用来从channel中读取客户端Socket发送过来的数据
Reader reader = getReader();
// 创建一个Connection对象，在register里把这个对象添加到ConnectionManager#connections集合里
Connection c = connectionManager.register(channel);
// If the connectionManager can't take it, close the connection.
// 如果超过了设置的最大连接数c == null
if (c == null) {
if (channel.isOpen()) {
IOUtils.cleanup(null, channel);
}
connectionManager.droppedConnections.getAndIncrement();
continue;
}
// 将Connection对象attach到此key上，这样后续通过key.attachment()即可拿到c这个对象
key.attach(c); // so closeCurrentConnection can get the object
// 把连接添加到Reader的Connection阻塞队列中
reader.addConnection(c);
}
}
接着我们来看Reader的run方法，前面我们看到了，在Listener类构造方法执行的时候，会初始化它的Reader[]成员变量readers，然后启动Reader线程，如下图：
所以接下来我们要看Reader的run方法：
很简单调用了doRunLoop方法，直接去看doRunLoop方法：
看下doRead方法，此方法中有两个地方需要深入研究：
① 通过key.attachment()拿到之前attach在此key的object，这个object是什么？又是什么时候attach到key上的的？
② readAndProcess函数的处理逻辑，看名字此函数就是读取客户端RPC请求然后进行处理的地方。
void doRead(SelectionKey key) throws InterruptedException {
int count;
// 从SelectionKey中拿出之前附着的Connection对象,一会我们再讲什么时候附着的
Connection c = (Connection)key.attachment();
if (c == null) {
return;
}
// 设置最近一次连接时间
c.setLastContact(Time.now());
try {
// 读客户端的RPC请求数据，并处理。这个函数后面分析
count = c.readAndProcess();
} catch (InterruptedException ieo) {
LOG.info(Thread.currentThread().getName() + ": readAndProcess caught InterruptedException", ieo);
throw ieo;
} catch (Exception e) {
// Any exceptions that reach here are fatal unexpected internal errors
// that could not be sent to the client.
LOG.info(Thread.currentThread().getName() +
": readAndProcess from client " + c +
" threw exception [" + e + "]", e);
count = -1; //so that the (count < 0) block is executed
}
// setupResponse will signal the connection should be closed when a
// fatal response is sent.
if (count < 0 || c.shouldClose()) {
closeConnection(c);
c = null;
}
else {
c.setLastContact(Time.now());
}
}
首先来回答第一个问题：通过key.attachment()拿到之前attach在此key的object，这个object是什么？又是什么时候attach到key上的的？
请看doRead方法前面的register方法，如下图所示：
看下register方法：register(Selector sel, int ops,Object att)
这个方法是NIO库的方法。用来将channel注册到给定的selector上，并设置监听事件类型为ops，同时在key上attach一个对象att。
接着看第二个问题：readAndProcess的逻辑：
文字描述一下：
主要是通过channel的read方法将流中数据读到ByteBuffer类型的成员变量中，ByteBuffer成员变量是在Connection类中，Connection类有几个不同作用的ByteBuffer，分别是data、dataLengthBuffer、connectionHeaderBuf、unwrappedData、unwrappedDataLengthBuffer。根据Hadoop的协议头，读取相应位置的数据到对应的ByteBuffer中，有关请求体的数据被读到data这个ByteBuffer中，然后调用processOneRpc处理这个请求，processOneRpc方法不是本文关注的重点，但是可以稍微描述一下他的作用：将RPC请求封装成Call，加入到Call Queue中，等待Server#Handler内部类去消费。