概述

在《Scalable IO in Java》中讲到了一种多线程下的Reactor模式。在这个模式里，mainReactor只有一个，负责响应client的连接请求，并建立连接，它使用一个NIO Selector；subReactor可以有一个或者多个，每个subReactor都会在一个独立线程中执行，并且维护一个独立的NIO Selector。

这样的好处很明显，因为subReactor也会执行一些比较耗时的IO操作，例如消息的读写，使用多个线程去执行，则更加有利于发挥CPU的运算能力，减少IO等待时间。

Netty 3.7中Reactor的EventLoop在AbstractNioSelector.run()中，它实现了Runnable接口。这个类是Netty NIO部分的核心。它的逻辑非常复杂，其中还包括一些对JDK Bug的处理（例如rebuildSelector），刚开始读的时候不需要深入那么细节。我精简了大部分代码，保留主干如下：

abstract class AbstractNioSelector implements NioSelector { //NIO Selector protected volatile Selector selector; //内部任务队列 private final Queue<Runnable> taskQueue = new ConcurrentLinkedQueue<Runnable>(); //selector循环 public void run() { for (;;) { try { //处理内部任务队列 processTaskQueue(); //处理selector事件对应逻辑 process(selector); } catch (Throwable t) { try { Thread.sleep(1000); } catch (InterruptedException e) { // Ignore. } } } } private void processTaskQueue() { for (;;) { final Runnable task = taskQueue.poll(); if (task == null) { break; } task.run(); } } protected abstract void process(Selector selector) throws IOException; }

其中process是主要的处理事件的逻辑，例如在AbstractNioWorker中，处理逻辑如下：

protected void process(Selector selector) throws IOException { Set<SelectionKey> selectedKeys = selector.selectedKeys(); if (selectedKeys.isEmpty()) { return; } for (Iterator<SelectionKey> i = selectedKeys.iterator(); i.hasNext();) { SelectionKey k = i.next(); i.remove(); try { int readyOps = k.readyOps(); if ((readyOps & SelectionKey.OP_READ) != 0 || readyOps == 0) { if (!read(k)) { // Connection already closed - no need to handle write. continue; } } if ((readyOps & SelectionKey.OP_WRITE) != 0) { writeFromSelectorLoop(k); } } catch (CancelledKeyException e) { close(k); } if (cleanUpCancelledKeys()) { break; // break the loop to avoid ConcurrentModificationException } } }

可以看到的是 和NIO的编程没有什么不同 都是通过遍历selelctKeys来处理准备就绪的IO 在Netty 4.0之后，作者觉得NioSelector这个叫法，以及区分NioBoss和NioWorker的做法稍微繁琐了点，干脆就将这些合并成了NioEventLoop，从此这两个角色就不做区分了。我倒是觉得新版本的会更优雅一点。

1 netty中的多线程

一旦对应的Boss或者Worker启动，就会分配给它们一个线程去一直执行。对应的概念为BossPool和WorkerPool。对于每个NioServerSocketChannel，Boss的Reactor有一个线程，而Worker的线程数由Worker线程池大小决定，但是默认最大不会超过CPU核数*2，当然，这个参数可以通过NioServerSocketChannelFactory构造函数的参数来设置。

public NioServerSocketChannelFactory( Executor bossExecutor, Executor workerExecutor, int workerCount) { this(bossExecutor, 1, workerExecutor, workerCount); }

我们之前ChannlePipeline中的ChannleHandler是在哪个线程执行的呢？答案是在Worker线程里执行的，并且会阻塞Worker的EventLoop。例如，在NioWorker中，读取消息完毕之后，会触发MessageReceived事件，这会使得Pipeline中的handler都得到执行。

protected boolean read(SelectionKey k) { …

if (readBytes > 0) { // Fire the event. fireMessageReceived(channel, buffer); } return true; }