通过Hadoop RPC框架学习NIO

Posted 2021-04-23 Java不睡觉

今儿个我们通过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内部类去消费。