多线程下的Reactor以及在netty中的应用

在上一篇文章讲解了Reactor单线程模型 本章讲一下其多线程模型及在netty的应用场景，由Reactor出发来理解Netty

1、多线程下的Reactor

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

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

2、Netty中的Reactor与NIO

好了，了解了多线程下的Reactor模式，我们来看看Netty吧（以下部分主要针对NIO，OIO部分更加简单一点，就不详细介绍了）。Netty里对应mainReactor的角色叫做“Boss”，而对应subReactor的角色叫做"Worker"。Boss负责分配请求，Worker负责执行，好像也很贴切！以TCP的Server端为例，这两个对应的实现类分别为NioServerBoss和NioWorker（Server和Client的Worker没有区别，因为建立连接之后，双方就是对等的进行传输了）。

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
 }
 }
 }

这不就是selector经典用法了么？

在4.0之后，作者觉得NioSelector这个叫法，以及区分NioBoss和NioWorker的做法稍微繁琐了点，干脆就将这些合并成了NioEventLoop，从此这两个角色就不做区分了。我倒是觉得新版本的会更优雅一点。

3、Netty中的多线程

下面我们来看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;
 }