EventLoop（netty源码死磕4）
精进篇：netty源码 死磕4-EventLoop的鬼斧神工
目录
1. EventLoop的鬼斧神工
2. 初识 EventLoop
3. Reactor模式回顾
3.1. Reactor模式的组成元素：
3.2. Reactor模式的三步曲
4. Netty中的Reactor模式应用
5. channel系列类结构
5.1. channel家族成员
5.2. NioSocketChannel 类的层次机构
5.3. netty channel 和本地Channel的关系
6. NioEventLoop
1.1. 和本地Selector的对应关系
1.2. 和Netty Channel的关系
7. Reactor三步曲之注册
1.1. Netty中Channel注册流程总览
1.2. 注册流程的关键代码
1.3. 第三个参数有机关
8. Reactor三步曲之轮询
1.1. Netty中EventLoop轮询流程总览
1.2. EventLoop线程启动
1.3. NioEventLoop 事件轮询
1.4. 取得就绪事件的个数
1.5. 就绪事件的迭代处理
1.6. processSelectedKey
9. Reactor三步曲之分派
1.1. dispatch（分派）结果
10. 总结
1. 初识 EventLoop
阅读netty的源码，首先从最为核心的、也是最为基础的EventLoop系列类入手。EventLoop 系列类，就像netty这座大厦的钢筋混凝土框架，是非常重要的基础设施。弄清楚EventLoop 的原理，是研读和学习netty架构的前提。
EventLoop 不是Netty中的一个类，而是一系列的类，或者说一组类。这一组类的作用，对应于Reactor模式的Reactor 角色。
呵呵，又回到了非常牛逼的反应器模式。
Reactor模式，是高性能JAVA编程的必知必会模式。首先熟悉Reactor模式，一定是磨刀不误砍柴工。
如果对Reactor模式还不太了了解，可以翻阅《基础篇：netty源码 死磕3-传说中神一样的Reactor反应器模式》，此文站在巨人的肩膀上，对Reactor模式做了极为详尽的介绍。
2. Reactor模式回顾
为了更好的展开陈述，还是简单的总结一下Reactor模式。
1.1. Reactor模式的组成元素：
channel和selector属于 java.nio 包中的类，分别为网络通信中的通道（连接）和选择器。
Reactor和 handler 属于Reactor模型高性能编程中的应用程序角色，分别为反应器和处理器。
1.2. Reactor模式的三步曲
从开发或者执行流程上，Reactor模式可以被清晰的被分成三大步：注册、轮询、分发。
第一步：注册
将channel 通道的就绪事件，注册到选择器Selector。在文章《基础篇：netty源码 死磕3-传说中神一样的Reactor反应器模式》的例子中，这块注册的代码，放在Reactor的构造函数中完成。一般来说，一个Reactor 对应一个选择器Selector，一个Reactor拥有一个Selector成员属性。
第二步：轮询
轮询的代码，是Reactor重要的一个组成部分，或者说核心的部分。轮询选择器是否有就绪事件。
第三步：分发
将就绪事件，分发到事件附件的处理器handler中，由handler完成实际的处理。
总体上，Netty是基于Reactor模式实现的，对于就绪事件的处理总的流程，基本上就是上面的三步。
3. Netty中的Reactor模式应用
下面进行Netty的Reactor模型和经典Reactor的对照说明。
Netty的Reactor模型，和经典Reactor模型的元素对应关系如下图：
Netty中的Channel系列类型，对应于经典Reactor模型中的client， 封装了用户的通讯连接。
Netty中的EventLoop系列类型，对应于经典Reactor模型中的Reactor，完成Channel的注册、轮询、分发。
Netty中的Handler系列类型，对应于经典Reactor模型中的Handler，不过Netty中的Handler设计得更加的高级和巧妙，使用了Pipeline模式。这块非常精彩，后面专门开文章介绍。
总之，基本上一一对应。所以，如果熟悉经典的Reactor模式，学习Netty，会比较轻松。
4. channel系列类结构
1.3. channel家族成员
Netty 还支持非常多的通讯连接协议，每种协议还有 NIO(异步 IO) 和 OIO(Old-IO， 即传统的阻塞 IO) 版本的区别。对应于不同协议，都有不同的 Channel 类型与之对应。
下面是一些常用的 Channel 类型:
NioSocketChannel： 代表异步的客户端 TCP Socket 连接.
NioServerSocketChannel： 异步的服务器端 TCP Socket 连接.
NioDatagramChannel： 异步的 UDP 连接
NioSctpChannel： 异步的客户端 Sctp 连接.
NioSctpServerChannel： 异步的 Sctp 服务器端连接.
OioSocketChannel： 同步的客户端 TCP Socket 连接.
OioServerSocketChannel： 同步的服务器端 TCP Socket 连接.
OioDatagramChannel： 同步的 UDP 连接
OioSctpChannel： 同步的 Sctp 服务器端连接.
OioSctpServerChannel： 同步的客户端 TCP Socket 连接.
以当下的编程来说，一般来说，用到最多的通讯协议，还是 TCP 协议。所以，本文选取NioSocketChannel 类，作为channel 连接家族类的代表，进行讲解。
了解了NioSocketChannel 类，在使用的方法上，其他的通道类型，基本上是一样的。
1.4. NioSocketChannel 类的层次机构
1.5. 和本地Channel的关系
阅读NioSocketChannel 源码，在其父类AbstractNioChannel中找到了一个特殊的成员属性ch，这个成员的类型是 java本地类型SelectableChannel 。在《Java NIO Channel (netty源码死磕1.3)》一文中已经讲到过，SelectableChannel 类型这个是所有java本地非阻塞NIO 通道类型的父类。
private final SelectableChannel ch; protected final int readInterestOp; volatile SelectionKey selectionKey;
也就是说，一个Netty Channel 类型，封装了一个java非阻塞NIO 通道类型成员。这个被封装的本地Java 通道成员ch，在AbstractNioChannel的构造函数中，被初始化。
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
try {
ch.configureBlocking(false);
} catch (IOException e) {
try {
ch.close();
} catch (IOException e2) {
if (logger.isWarnEnabled()) {
logger.warn(
"Failed to close a partially initialized socket.", e2);
}
}
throw new ChannelException("Failed to enter non-blocking mode.", e);
}
}因为是非阻塞IO， ch.configureBlocking（false） 方法被调用，通道被设置为非阻塞。
5. NioEventLoop
终于到了重点。
Netty中的NioEventLoop类，就是对应于非阻塞IO channel的Reactor 反应器。
1.6. NioEventLoop 和本地Selector的对应关系
NioEventLoop类型绑定了两个重要的java本地类型：一个线程类型，一个Selector类型。
本地Selector属性的作用，用于注册Java本地channel。本地线程属性的作用，主要是用于轮询。
NioEventLoop 源码如下
public final class NioEventLoop extends SingleThreadEventLoop {
………………….
/**
* The NIO {@link Selector}.
*/
private final SelectableChannel ch;
protected final int readInterestOp;
volatile SelectionKey selectionKey;
…………………..
}
在父类SingleThreadEventExecutor 中，定义了一个线程属性thread，源码如下（省略了不相干的内容）：
public abstract class SingleThreadEventExecutor ….{
………………….
private final Thread thread;
private final ThreadProperties threadProperties;
…………………..
}
线程什么时候启动呢？
在reactor模式中，线程是轮询用的。所以，Reactor线程的启动，一般在channel的注册之后。
1.7. EventLoop 和Netty Channel的关系
Netty中，一个EventLoop，可以注册很多不同的Netty Channel。相当于是一对多的关系。
这一点，和Java NIO中Selector和channel的关系，也是一致的。
通过上面的分析，我们已经知道了Netty 的非阻塞IO，是建立在Java 的NIO基础之上的。
如果对Java 的NIO不了解，请阅读下面的四文：
《Java NIO Buffer(netty源码死磕1.2)》
《Java NIO Channel (netty源码死磕1.3)》
《Java NIO Selector (netty死磕1.4)》
6. Reactor三步曲之注册
对于Java NIO而言，第一步首先是channel到 seletor的事件就绪状态的注册。对于Netty而言，也是类似的。
在此之前，Netty有一些启动的工作需要完成。这些启动的工作，包含了EventLoop、Channel的创建。 这块BootStrap 的启动类和系列启动工作，后面有文章专门介绍。下面假定启动工作已经完成和就绪，开始进行管道的注册。
1.8. Netty中Channel注册流程总览
Channel向EventLoop注册的过程，是在启动时进行的。注册的入口代码，在启动类AbstractBootstrap.initAndRegister 方法中。
注册入口代码如下：
final ChannelFuture initAndRegister() {
// .........
final Channel channel = channelFactory().newChannel();
init(channel);
ChannelFuture regFuture = group().register(channel);
}
完整的注册调用流程如下：
1.9. 注册流程的关键代码
关键代码如下，主要在AbstractChannel类中：
public abstract class AbstractChannel ….{
// 内部类AbstractUnsafe
protected abstract class AbstractUnsafe implements Unsafe {
//倒数第三步
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
………………….
if (eventLoop.inEventLoop()) {
register0(promise);
}else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
………………….
//倒
private void register0(ChannelPromise promise) {
boolean firstRegistration = neverRegistered;
neverRegistered = false;
registered = true;
safeSetSuccess(promise);
pipeline.fireChannelRegistered();
// Only fire a channelActive if the channel has never been registered. This prevents firing
// multiple channel actives if the channel is deregistered and re-registered.
if (firstRegistration && isActive()) {
pipeline.fireChannelActive();
}
}
…………………..
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
在倒数的第一步，也就是最后一步中，Netty的Channel通过javaChannel()方法，取得了Java本地Channel。
这个javaChannel()方法，它返回的是一个 Java NIO SocketChannel。
前面讲到，AbstractNioChannel通道类有一个本地Java 通道成员ch，在AbstractNioChannel的构造函数中，被初始化。 javaChannel()取到的，就是这个ch成员属性。通过最后一步，Netty终于将这个 SocketChannel 注册到与 eventLoop 关联的 selector 上了。
在注册的倒数第三步：
AbstractChannel.this.eventLoop = eventLoop;
在这个 AbstractChannel#AbstractUnsafe.register 中，会将一个 EventLoop 赋值给 AbstractChannel 内部的 eventLoop 字段, 到这里就完成了 EventLoop 与 Channel 的关联过程.
反过来说：这一句，将 Channel 与对应的 EventLoop 关联和绑定，也就是说， 每个 Channel 都会关联一个特定的 EventLoop。
在关联好 Channel 和 EventLoop 后, 会继续调用底层的 Java NIO SocketChannel 的 register 方法, 将底层的 Java NIO SocketChannel 注册到指定的 selector 中。
庆祝下， 通过这两步, 就完成了 Netty Channel 的注册过程。 从上到下，全部关联了哈。
1.10. 第三个参数有机关
到了这里，先别太捉急高兴。
再回到最后一步，看一下AbstractChannel 的doRegister() 代码。
//倒数第一步，执行注册
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
return;
} catch (CancelledKeyException e) {
.....
}
}
}
}
特别注意一下 register 的第三个参数，这个参数是设置 selectionKey 的附加对象的， 和调用 selectionKey.attach(object) 的效果一样。
下面是经典Reactor模式的注册代码：
Reactor(int port) throws IOException
{ //Reactor初始化
selector = Selector.open();
serverSocket = ServerSocketChannel.open();
serverSocket.socket().bind(new InetSocketAddress(port));
//非阻塞
serverSocket.configureBlocking(false);
//分步处理,第一步,接收accept事件
SelectionKey sk =
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
//attach callback object, Acceptor
sk.attach(new Acceptor());
}这段经典Reactor模式代码中，调用二个参数的register 方法，然后再附加对象。这种分离附加对象的方式，和前面调用三个参数的register 方法，结果是一样的。
再回到最后一步，看一下AbstractChannel 的doRegister() 代码。
doRegister（）所传递的第三个参数是 this,，它就是一个 NioSocketChannel 的实例。简单的说——Netty将 SocketChannel 对象自身，以附加字段的方式添加到了selectionKey 中，供事件就绪后使用。
后面会怎么样使用这个附加字段呢？
且看Reactor三步曲之二——轮询。
7. Reactor三步曲之轮询
1.11. Netty中EventLoop轮询流程总览
EventLoop 作为Reactor反应器的角色，是Reactor模式的核心。在Channel注册完成之后，EventLoop 就会开启轮询模式。
整个轮询的过程，和经典的Reactor模式的流程大致相同。在Netty中分为以下四步。
在讲解轮询的流程前，首先介绍一下轮询线程的启动。
1.12. EventLoop线程启动
前面讲到，Netty中，一个 NioEventLoop 本质上是和一个特定的线程绑定, 这个线程保存在EvnentLoop的父类属性中。
在EvnentLoop的父类SingleThreadEventExecutor 中，有一个 Thread thread 属性, 存储了一个本地 Java 线程。
线程在哪里启动的呢？
细心的你，有可能在前面已经发现了。
在前面的倒数第三步的注册中，函数 AbstractChannel.AbstractUnsafe.register中，有一个eventLoop.execute（）方法调用，这个调用，就是启动EvnentLoop的本地线程的的入口。
重复贴一次，代码如下：
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}在execute的方法中，去调用 startThread()，启动线程。
代码如下：
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
// 调用 startThread 方法, 启动EventLoop 线程.
startThread();
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
SingleThreadEventExecutor.startThread() 方法中了:
private void startThread() {
if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
thread.start();
}
}
}
终于看到朝思暮想的线程启动方法了。
它既是：thread.start()
STATE_UPDATER 是 SingleThreadEventExecutor 内部维护的一个属性, 它的作用是标识当前的 thread 的状态。在初始的时候, STATE_UPDATER == ST_NOT_STARTED, 因此第一次调用 startThread() 方法时, 就会进入到 if 语句内, 进而调用到 thread.start().
1.13. NioEventLoop 事件轮询
事件的轮询，在NioEventLoop.run() 方法, 其源码如下:
@Override
protected void run() {
for (;;) {
boolean oldWakenUp = wakenUp.getAndSet(false);
try {
//第一步，查询 IO 就绪
if (hasTasks()) {
selectNow();
} else {
select(oldWakenUp);
if (wakenUp.get()) {
selector.wakeup();
}
}
//第二步，处理这些 IO 就绪
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
processSelectedKeys();
runAllTasks();
} else {
final long ioStartTime = System.nanoTime();
processSelectedKeys();
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
break;
}
}
} catch (Throwable t) {
...
}
}
}
完成第二步IO就绪事件处理的调用是processSelectedKeys() ，这个调用非常关键。
这个方法是查询就绪的 IO 事件, 然后处理它；第二个调用是 runAllTasks(), 这个方法的功能就是运行 taskQueue 中的任务。
关于EventLoop中如何处理任务，后面用专门的文章来讲解。
1.14. 取得就绪事件的个数
void selectNow() throws IOException {
try {
selector.selectNow();
} finally {
// restore wakup state if needed
if (wakenUp.get()) {
selector.wakeup();
}
}
}
首先调用了 selector.selectNow() 方法，这个 selector 属性正是 Java NIO 中的多路复用器 Selector。selector.selectNow() 方法会检查当前是否有就绪的 IO 事件。如果有, 则返回就绪 IO 事件的个数；如果没有, 则返回0。
注意, selectNow() 是立即返回的，不会阻塞当前线程。 当 selectNow() 调用后， finally 语句块中会检查 wakenUp 变量是否为 true，当为 true 时, 调用 selector.wakeup() 唤醒 select() 的阻塞调用。
1.15. 就绪事件的迭代处理
private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
for (int i = 0;; i ++) {
final SelectionKey k = selectedKeys[i];
if (k == null) {
break;
}
selectedKeys[i] = null;
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
...
}
}
迭代 selectedKeys 获取就绪的 IO 事件, 然后为每个事件都调用 processSelectedKey 来处理它.
在前面的channel注册时，将 SocketChannel 所对应的 NioSocketChannel 以附加字段的方式添加到了selectionKey 中。
在这里， 通过k.attachment()取得这个通道对象，然后就调用 processSelectedKey 来处理这个 IO 事件和通道。
1.16. processSelectedKey
private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final NioUnsafe unsafe = ch.unsafe();
...
try {
int readyOps = k.readyOps();
// 读就绪
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
if (!ch.isOpen()) {
// Connection already closed - no need to handle write.
return;
}
}
// 写就绪
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
// 连接建立就绪事件
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
........
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
上面的代码，已经回到经典Reactor模式了。
processSelectedKey 中处理了三个事件, 分别是:
OP_READ, 可读事件, 即 Channel 中收到了新数据可供上层读取.
OP_WRITE, 可写事件, 即上层可以向 Channel 写入数据.
OP_CONNECT, 连接建立事件, 即 TCP 连接已经建立, Channel 处于 active 状态.
8. Reactor三步曲之分派
1.17. dispatch（分派）结果
在AbstractNioByteChannel 中，可以找到 unsafe.read( ) 调用的实现代码。 unsafe.read( )负责的是 Channel 的底层数据的 IO 读取，并且将读取的结果，dispatch（分派）给最终的Handler。
AbstractNioByteChannel.read（）的关键源码节选如下：
@Override
public final void read() {
...
ByteBuf byteBuf = null;
int messages = 0;
boolean close = false;
try {
int totalReadAmount = 0;
boolean readPendingReset = false;
do {
// 读取结果.
byteBuf = allocHandle.allocate(allocator);
int writable = byteBuf.writableBytes();
int localReadAmount = doReadBytes(byteBuf);
...
// dispatch结果到Handler
pipeline.fireChannelRead(byteBuf);
byteBuf = null;
...
totalReadAmount += localReadAmount;
...
}
}
9. 总结
到此为止，EventLoop的整个流程，已经分析完了
下一篇文章，将解读Netty的Handler。
无编程不创客，无案例不学习。疯狂创客圈，一大波高手正在交流、学习中!
疯狂创客圈 Netty 死磕系列 10多篇深度文章： 【博客园 总入口】 QQ群：104131248