作者：星巴刻
前文《 NioEventLoop 的职责》从外部的角度阐述了 NioEventLoop 的首要责任：为注册在 NioEventLoop 上的 channels 服务，发现 channels 上的 I/O 事件，然后将事件转交 channel 流水线 pipeline 处理。
本文是它的后续，剖析 NioEventLoop 的内部逻辑以及周边的一些代码，阐述 NioEventLoop 如何履行上述责任。
一、NioEventLoop 的继承体系
下图清晰地呈现了 NioEventLoop 的继承体系:
1. NioEventLoop的继承体系
图中呈现了 NioEventLoop 类的继承层级，虽然没有呈现它实现了哪些接口，不过不妨碍对这个图进行一些解读：
1、NioEventLoop 可以被追溯到 java.util.concurrent.ExecutorService 接口。这个信息表明 NioEventLoop 能够执行 Runnable 任务，从而保证了这些要执行的任务是在 NioEventLoop 的线程中执行，而非外部线程来执行。NioEventLoop 内置的一些 Runnable 任务包括了对 channel 的 register、系统缓冲区满后而推迟的 flush 任务等。
2、NioEventLoop 继承于 io.netty.util.concurrent.SingleThreadEventExecutor。这个信息表明，NioEventLoop 是单线程的。根据 SingleThreadEventExecutor 的说明，提交到 SingleThreadEventExecutor 的任务，将按照提交的先后顺序执行。
需要注意的是，开发不要随便用 NioEventLoop 执行长任务以免造成 I/O 阻塞，给 NioEventLoop 履行其首要职责添堵。更不能在 NioEventLoop 执行含有类似 Thread.sleep()、Object.wait()、ReentrantLock.lock()、CountDownLatch.await() 等等之类的程序，尽量避免使用 synchronized 同步锁，这些都可能导致 NioEventLoop 线程中断（线程这口气要是断了，I/O 的命就没了）。
二、SingleThreadEventExecutor 简述
io.netty.util.concurrent.SingleThreadEventExecutor 内部包含了一个任务 FIFO 队列，用于存储提交到该类的 Runnable 任务。它也创建了一个定制的单线程对象，规定了线程执行之后结束之前的一些处理，比如把未执行完成的任务执行完毕、关闭 NioEventLoop 的 Selector 等。它提供了addTask、offerTask、removeTask、runAllTasks等仅供子类可以调用的方法。但没有具体实现线程具体要执行的内容，它空出了 run 方法让子类来实现，所以 SingleThreadEventExecutor 类本身默认不会从它的任务队列中取出任务进行执行：
protected abstract void run();
作为子类的 NioEventLoop 将实现这个方法，它也就成了 NioEventLoop 的重要入口方法。NioEventLoop 的 run 方法实现后，将被 SingleThreadEventExecutor 里创建的线程执行， NioEventLoop 所负的对 channels 的职责，将在 run 方法中实现。
三、NioEventLoop 的创建
一般，应用程序开发时并不会直接创建 NioEventLoop，而是通过创建 NioEventLoopGroup 提供给 Netty 的 Bootstrap 或 ServerBootstrap。所以 NioEventLoop 并没有 public 的构建方法，它的创建必须由与它同一个 package 下的 NioEventLoopGroup 负责。简单地说，NioEventLoopGroup 可以认为是 NioEventLoop 的工厂。NioEventLoopGroup 的 next() 方法是提供 NioEventLoop 的工厂方法，用于返回一个提前创建好的 NioEventLoop。NioEventLoopGroup 提供了多个构造方法，构造 NioEventLoopGroup 时，可以指定 nThreads 参数表示这个 NioEventLoopGroup 要创建多少个 NioEventLoop；如果不指定 nThreads或指定为 0，则默认创建 2 倍 CPU 核数的 NioEventLoop。NioEventLoop 的获取的示例代码如下：
NioEventLoopGroup gorup = new NioEventLoopGroup(4);
NioEventLoop loop = group.next();
NioEventLoop 被创建时，会打开一个自己独有的 java.nio.channels.Selector。Selector 是 Java Nio 的选择器，NioEventLoop 之所以能够发现 channels 的 I/O 事件，正是使用了它。假如我是一个服务端程序，有一个 NioEventLoop 作为 parent， 有 16 个 NioEventLoop 作为 child，那么这个程序内部将由 1 + 16 = 17 的 NioEventLoop，每个 NioEventLoop 分别含有 1 个自己的 Selector，因此这个程序内部总共将含有 17 个 Selector。
四、将 Channel 注册到 NioEventLoop
为了侦听客户端连接的到来，需要将创建的 NioServerSocketChannel 注册到作为 parent 的 NioEventLoop 上；为了发现并处理客户端连接的可读可写事件，需要将表示客户端连接的 NioSocketChannel 注册到分配给它的、充当 child 角色的某个 NioEventLoop 上。这里要注意的是，充当 child 角色的 NioEventLoop 在程序里有多个，一个 NioEventLoop 可以负责多个客户端连接，但是每一个客户端连接只能被一个 NioEventLoop 负责。
NioEventLoop 的父类 SingleThreadEventLoop 提供了如下注册方法，可以用于注册 NioServerSocketChannel 或 NioSocketChannel：
code: SingleThreadEventLoop.register
以上代码片段表示，SingleThreadEventLoop.register(channel, promiss) 方法将注册调用转交给 channel 的 unsafe 对象的 register 方法完成，也就是说实际由 NioServerSocketChannel 或 NioSocketChannel 的 unsafe.register() 方法来具体完成注册工作。这两个类的 unsafe 对象刚好都在它们的父类 AbstractChanel 中定义和创建。在 AbstractUnsafe.register() 方法中，首先会判断当前执行 register 的线程是在负责该 channel 的 NioEventLoop 的内部线程中，如果是则继续，如果不是则会把接下来的工作封装为任务，提交到 NioEventLoop 的任务队列中，从而保证最终执行 register 操作的线程是由 NioEventLoop 的内部线程来执行。下面是 AbstractUnsafe 中关于注册的核心代码，如前所述，这个代码将由 NioEventLoop 的内部线程来执行：
code: AbstractUnsafe.register0
register0(promiss) 方法首先先调用 channel 的 doRegister() 方法，在 AbstractChanel 中，doRegister() 是一个抽象方法，作为 NioServerSocketChannel、NioSocketChannel 的父类的 AbstractNioChannel 类实现了 doRegister() 方法，它的实现体现了这里的本质：取出 NioServerSocketChannel、NioSocketChannel 底层的 java.nio.channels.SelectableChannel 对象，调用它提供 register 方法，将 SelectableChannel 注册到负责 NioServerSocketChannel、NioSocketChannel 的 NioEventLoop 所创建的 selector 上：
code: AbstractNioChannel.doRegister
有点意思的是，调用 javaChannel().register() 方法，传入的 interestOps（第 2 个参数）此时并不指定任何值，只是简单地传入 0，这说明 Netty 对 channel I/O 事件兴趣的注册并不是在注册时设置的。
回到 "code: AbstractUnsafe.register0"，程序调用完 channel.doRegister 后，关联的 pipleline 有 2 个重要方法被调用到: pipleline.fireChannelRegistered 和 pipleline.fireChannelActive 。从而将 channel 的注册事件、就绪事件传递到 pipeline 中，应用程序可以可以处理了。
此处也可以看出，channel 第一次的 register 状态将先于 active 状态送到 pipleline。但如果 channel 后续取消注册后的再次注册，register 会继续发送给 pipeline，而 active 则不会。
pipleline.fireChannelRegistered 发送后，注册到 pipeline 的 ChannelInboutHandler 的 channelRegistered(ctx)将被调用。
但 pipleline.fireChannelActive 发送后，注册到 pipeline 的 ChannelInboutHandler，不仅仅 channelActive(ctx) 会被调用到，默认情况下还会修改该 channel 的 selectionKey 的 interestOps值，使得当有信息到来时 channelRead(ctx) 也会被调用到（除非 channel.config.isAutoConfig 配置为 false 了）; 这个逻辑由 DefaultChannelPipeline 实现：
code: DefaultChannelPipeline.HeadContext.channelActive
上图中 readIfIsAutoRead() 方法调用了 channel.read() 方法，通过一系列迂回，最终将调用到 AbstractNioChannel 的 doBeginRead 方法，在这个方法中修改 selectionKey 的 interestOps，增加 readInterstOp 操作（对 NioServerSocketChannel 而言 readInterstOp 指的是 OP_ACCEPT，对 NioSocketChannel 而言，readInterstOp 指的是 OP_READ)，表明程序可以开始处理读请求:
code: AbstractNioChannel.doBeginRead
五、通过 Selector 发现 I/O 事件，并转交给 pipeline 处理
假设此时 NioEventLoop 已经被注册了 1 个或多个 Channel 了。现在回到 NioEventLoop 实现 SingleThreadEventExecutor 要求的 run 方法的地方，run 方法大致的流程如下：
1. 通过 selector.select 或 selectNow 方法从 Selector 中发现当前发生的 I/O 事件.
select 方法是阻塞的，selectNow 是非阻塞的。NioEventLoop 内部会进行决策，什么时候使用阻塞，什么时候使用非阻塞。简单地，如果 NioEventLoop 中当前的任务队列中有任务时候，会使用 selectNow 非阻塞；如果当前任务队列中没有任务，则会优先进行一次非阻塞 selectNow，如果没有新的 I/O 事件，则再使用阻塞的 select。
2. 调用 selector.select 阻塞方法时候，会传入一个超时时间
如果不传入一个超时时间，selector.select 会一直阻塞下去直到由 I/O 事件的到来。如果真这样，那这个期间线程将被挂起，注册到 NioEventLoop 中的定时任务（比如Idle检测任务）就得不到执行。所以 Netty 调用阻塞 select 方法之前，会根据当前定时任务情况，计算出一个可以等待 I/O 事件的最迟时间，以便恢复线程。
当有 I/O 事件处理时，Netty 处理这些事件的顺序是一个一个 channel 处理，对每个 channel 的事件处理顺序是，先处理 OP_CONNECT、后处理 OP_WRITE，最后处理 OP_READ 和 OP_ACCEPT。这里有 2 个细节：
一，优先处理 OP_CONNECT 事件。优先处理 OP_CONNECT 事件是因为 Java Nio 要求，对于客户端新建立的连接，必须先完成对 channel 的 finishConnect 调用后，才能使用该 channel，否则会抛出 NotYetConnectedException 异常；
二，把处理 OP_WRITE 的优先级提高到比 OP_READ、OP_ACCEPT前，目的是为了希望先处理 OP_WRITE 腾出一些堆内存空间，使得有更多的空闲内容能够为OP_READ、OP_ACCEPT 服务。
3. OP_CONNET 事件的处理
OP_CONNET 事件在客户端程序中出现，代表底层已经和远程建立好连接了。根据 Java Nio 的要求，此时需要调用 java.nio.channels.SocketChannel 的 finishConnect() 方法，完成连接的建立。NioSocketChannel 的 doFinishConnect 完成这个工作。
对应用程序而言，此时连接已经建立，就要进入 active 状态。pipeline 的 channelActive(ctx) 要被调用到。AbstractNioChannel.AbstractNioUnsafe 的 fulfillConnectPromise 方法完成这个工作。
最后，对于一个连接而言，OP_CONNECT 只会发生一次，所以一旦接收到 OP_CONNECT 事件后，就可以取消对这类事件的关注了，通过 selectionKey.interestOps(k.interestOps() &~SelectionKey.OP_CONNECT) 完成这个工作。
4. OP_WRITE 事件的处理
OP_WRITE 事件表示当前 channel 的系统缓冲区有空，应用程序可以往系统缓冲区写入信息。Netty 对 OP_WRITE 事件的响应就是调用 channel.unsafe().forceFlush()。
一般情况下，系统缓冲区都不会满，除非是瞬间发送大量的信息或者客户端接收速度相对较慢。因此，Netty 默认情况下，不会向 selectionKey 注册对 OP_WRITE 感兴趣，否则每次调用 selector.select 将因为具备 OP_WRITE 条件立即返回而不会阻塞，导致大量 CPU 被浪费。
应用程序调用 channel.write(msg)，并不会把信息直接写入系统（发送）缓冲区，而是先把要写入的信息写入专属于 channel 的 ChannelOutboundBuffer（一个 channel 实例有一个专属的 ChannelOutboundBuffer 实例）。只有当调用 channel.flush() 或 ctx.flush() 时，Netty 才会真正尝试把信息写入系统缓冲区。在 NioSocketChannel.doWrite(ChannelOutboundBuffer) 方法尝试将 ChannelOutboundBuffer 中的信息写入系统缓冲区时，如果系统缓冲区已满，Netty 无法写入任何一个字节到系统缓冲区，此时 doWrite 将调用 incompleteWrite(true) 才向该 channel 的 selectionKey 注册对 OP_WRITE 感兴趣。需要注意 Netty 这里有一个小技巧，即使无法全部写完，但只要 doWrite 能够往系统缓冲区写入哪怕一个字节，Netty 都不会向 selectionKey 注册 OP_WRITE，而是调用 incompleteWrite(false) 往任务队列的增加一个新的 flush() 任务，过一会再尝试进行 flush() 。这个小技巧体现了 Netty 对细节的把握：或许一会系统缓冲区就好了呢？
5、OP_READ、OP_ACCEPT 事件的处理
OP_READ 事件表示，当前系统缓冲区有数据需要应用程序来读取；OP_ACCEPT 事件表示，当前有新的客户端建立到服务端来了。这两个事件分别对应 NioSocketChannel 和 NioServerScoketChannel。Netty 把这个 I/O 事件抽象为 channel 有可读的事件。因此，Netty 直接调用各自 channel.unsafe 的 read() 方法直接处理：
对于 NioServerSocketChannel，channel.unsafe 由 AbstractNioMessageChannel 的NioMessageUnsafe 实现。unsafe.read() 方法将调用到 NioServerSocketChannel的 doReadMessages() 方法，从 Java Nio 的 Channel 中 accept 出一个 SocketChannel：
对于 NioSocketChannel, channel.unsafe 由 AbstractNioByteChannel 中的 NioByteUnsafe 实现，核心代码是最终调用到 NioSocketChannel 的 doReadBytes(ByteBuf) 方法，从 Java Nio 的 channel 中读取字节信息到 Netty 的 ByteBuf 中，然后调用 pipeline 的 fireChannelRead(ByteBuf)：
经过多次调用 doReadBytes(ByteBuf) 以及 pipeline.fireChannelRead(ByteBuf)，系统缓冲区的可读字节信息都读取完毕后 ，unsafe 的 read 方法会调用一次 pipeline.fireChannelReadComplete()。根据这个信息，这里有一个小技巧，尽量不要在 ChannelHandler 的 channelRead(ctx) 方法中调用 ctx 的 flush，而是在 pipeline.channelReadComplete(ctx) 再调用 flush，以减少过多的 flush 调用。
六、小花絮：SingleThreadEventExecutor 内部线程并不自动启动
SingleThreadEventExecutor 内部的线程不会自动启动，只有第一次接收外部任务时候才会自动启动!
为啥不默认启动呢，反正最终都要启动为何这么羞涩？这或许和 NioEventLoopGroup 构建时默认就把所有的 NioEventLoop 构建完毕有关，因为是提前构建但不一定实际使用，所以就先不 startThread 了。
下面这个代码是玩了一下一些特性的时候，需要特殊处理下：自己创建 Java Nio 的 Channel，然后注册到 Netty 的 NioEventLoop 中。需要自己写一个画红线部分的代码提交一个空方法来启动 NioEventLoop 内部的线程!
七、总结
NioEventLoop 比较繁杂，但它却是 Netty 内部的重要控制类，需要很好地弄懂它。
NioEventLoop 无法自己创建，必须由 NioEventLoopGroup 创建。多个 AbstractNioChannel 可以注册到一个 NioEventLoop 上。NioEventLoop 内部有一个专属的 Selector 对象以及一个线程。这个线程从 Selector 中 select 出感兴趣的 I/O 事件，调用 channel 的 unsafe 对象来处理。unsafe 负责按照 Java Nio 规范处理 I/O 事件，同时调用 channel.pipeline 的相关方法，使得应用程序提供的 ChannelHandler 能够得到回调。
2017-11-19
网友评论