版本
本次源码分析基于Netty的版本为4.1
源码分析
NioEventLoop可以视为java中的一个线程，只不过NioEventLoop处理的事件，以及内部的处理逻辑会有所不同。先看一下类的继承关系：
可以看到NioEventLoop实现了很多接口，特别是EventLoop和ScheduledExecutorService，所以NioEventLoop不仅能实现普通的task，还能实现定时task。
Selector
Netty的实现是基于Java原生的NIO的，其对原生的NIO做了很多优化，避免了某些bug，也提升了很多性能。但是底层对于网络IO事件的监听和处理也是离不开多路复用器Selector的。在NioEventLoop的构造方法中进行了Selector的初始化：
final SelectorTuple selectorTuple = openSelector();
selector = selectorTuple.selector;
关键还是openSelector()方法，这里我删除了一些分支代码，剩下的做了注释，其中常量 DISABLE_KEY_SET_OPTIMIZATION 的定义如下，可以手工配置是否开启优化，默认是开启优化的，具体优化做了什么事可以查看下面的openSelector()分析。
private static final boolean DISABLE_KEY_SET_OPTIMIZATION = SystemPropertyUtil.getBoolean("io.netty.noKeySetOptimization", false);
netty在创建selector的时候就尝试了优化，具体优化其实是将底层的数据结构从HashSet改为了数组，可以从SelectedSelectionKeySet和SelectorImpl的源码看到这一点，这里就不列了。
private SelectorTuple openSelector() {
final Selector unwrappedSelector;
try {
// 根据底层的IO模型来创建一个selector，这里的selector就是java中NIO的selector
unwrappedSelector = provider.openSelector();
} catch (IOException e) {
throw new ChannelException("failed to open a new selector", e);
}
// 如果未开启优化则直接就返回了，SelectorTuple可以视为一个持有selector引用的句柄
if (DISABLE_KEY_SET_OPTIMIZATION) {
return new SelectorTuple(unwrappedSelector);
}
Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {
@Override
public Object run() {
try {
// 通过反射创建一个selector的具体实例
return Class.forName(
"sun.nio.ch.SelectorImpl",
false,
PlatformDependent.getSystemClassLoader());
} catch (Throwable cause) {
return cause;
}
}
});
final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
// netty自己包装的一个selectKey的集合类
final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
@Override
public Object run() {
try {
Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) {
// java 9 以上版本会用Unsafe类直接底层替换SelectionKeySet
}
// 利用反射将原生selector中的两个属性替换为netty自己的包装类
selectedKeysField.set(unwrappedSelector, selectedKeySet);
publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
return null;
} catch (NoSuchFieldException e) {
return e;
} catch (IllegalAccessException e) {
return e;
}
}
});
selectedKeys = selectedKeySet;
return new SelectorTuple(unwrappedSelector, new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet));
}
run()方法
前面也说了，NioEventLoop其实其实可以类比java中的线程，是一个任务执行单元，所以run()方法是其中的关键，接下来就来分析一下run()方法，源码如下。
@Override
protected void run() {
for (;;) {
try {
try {
// 如果任务队列非空，那么执行selectNowSupplier代表的方法，也就是selectNow()，否则返回SelectStrategy.SELECT
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
// NioEventLoop不支持，用于EpollEventLoop，理论上不会走到这里
case SelectStrategy.BUSY_WAIT:
// 任务队列为空的时候，会执行本逻辑
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
// 源码中有注释为什么这里要调用如下逻辑，感兴趣可以查看源码，因为原因描述太长，这里就省略了
if (wakenUp.get()) {
selector.wakeup();
}
// fall through
default:
}
} catch (IOException e) {
// 出现IOException则新建selector，将原有的所有channel重新注册到新的selector，然后关闭老的selector
rebuildSelector0();
// 异常处理
handleLoopException(e);
continue;
}
cancelledKeys = 0;
needsToSelectAgain = false;
// 控制IO处理时间的一个变量，默认是50(代表50%)
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
// 处理IO事件
processSelectedKeys();
} finally {
// 运行非IO任务，就算ioRatio设置了100，非IO任务还是要执行的
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
// 处理IO事件
processSelectedKeys();
} finally {
// 根据设置的时间占比运行非IO任务
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
整个run()方法被包裹在一个for循环中，唯一能够结束循环的条件是状态state为SHUTDOWN或者TERMINATED，NioEventLoop继承了SingleThreadEventExecutor，isShuttingDown()和confirmShutdown()都是SingleThreadEventExecutor中的方法。
可以看到，除去异常处理和一些分支流程，整个run()方法不是特别负责，重点在与select()和selectNow()方法，run()方法流程如下图所示：
接下来看一下两个关键方法select()和selectNow()
- selectNow()
selectNow会立即出发selector的选择操作，如果有准备就绪的channel，就会返回相应的int值(代表了不同的selectKey的集合)，否则返回0。之后如果发现用户手动调用了selector的wakeup()方法，会执行selector.wakeup()操作。
int selectNow() throws IOException {
try {
return selector.selectNow();
} finally {
// restore wakeup state if needed
if (wakenUp.get()) {
selector.wakeup();
}
}
}
- select()
同样去掉了一些无关主流程的代码，netty在select()方法中的处理逻辑跟java线程池有相似的地方，没有任务的时候都是阻塞的，阻塞时间以最近的任务距离当前时间为准，如果一旦有就绪的channel，则立即进行退出循环进行处理。这里netty还解决了epoll的空轮询bug，如果触发了空轮询判断会重建selector。
private void select(boolean oldWakenUp) throws IOException {
Selector selector = this.selector;
try {
int selectCnt = 0;
long currentTimeNanos = System.nanoTime();
// 计算定时任务队列中最早的任务距离现在的时间，没有任务默认1秒
long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
for (;;) {
// 如果最早的任务开始时间距离当前时间不足0.5毫秒或者已超时，执行selectNow()方法
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
if (timeoutMillis <= 0) {
if (selectCnt == 0) {
selector.selectNow();
selectCnt = 1;
}
break;
}
// 队列中有任务，并且selector从false设置为true成功则执行selectNow()方法
// 源码描述了原因，简单来说，往NioEventLoop中提交任务的时候如果selector未wakeup会调用selector.wakeup()
// 但如果提交task的时候selector已经wakeup，则不会调用
// 任务可能被挂起知道selector超时，所以这里做了检测
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
selector.selectNow();
selectCnt = 1;
break;
}
// 为select方法设置超时，防止定时任务饿死
int selectedKeys = selector.select(timeoutMillis);
selectCnt ++;
// 退出循环的条件
// 1. 存在就绪的channel
// 2. 老的wakeup状态是true
// 3. 进入select方法后用户调用了wakeup()方法
// 4. 有新的定时任务需要处理
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
break;
}
if (Thread.interrupted()) {
// 线程中断处理
if (logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely because " +
"Thread.currentThread().interrupt() was called. Use " +
"NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
}
selectCnt = 1;
break;
}
long time = System.nanoTime();
// 这里有一个处理epoll空轮询bug的逻辑
// 超过了timeoutMillis时间不认为是空轮询
// 当select轮询超过设定的次数上限时视为触发空轮询bug，重建selector
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
// timeoutMillis elapsed without anything selected.
selectCnt = 1;
} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
// The code exists in an extra method to ensure the method is not too big to inline as this
// branch is not very likely to get hit very frequently.
selector = selectRebuildSelector(selectCnt);
selectCnt = 1;
break;
}
currentTimeNanos = time;
}
} catch (CancelledKeyException e) {
// 略
}
现在真的有channel就绪了，NioEventLoop会怎么处理呢？回到run()方法，有一段根据设定的时间比例处理IO事件和用户任务的逻辑，分别对应两个方法processSelectedKeys和runAllTasks
- processSelectedKeys()
从源码可以看到，processSelectedKeysOptimized和processSelectedKeysPlain的大部分处理逻辑是相同的，区别就在于对selectedKey的迭代逻辑，记得一开始说过如果开启了优化，netty对selectedKey的底层集合进行了优化，将HashSet改为了数组，HashSet底层用HashMap实现，迭代的效率是没有数组高的。
private void processSelectedKeys() {
// 看文章最开头是否启用优化的设置，如果启用了会走这里
if (selectedKeys != null) {
processSelectedKeysOptimized();
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
}
private void processSelectedKeysOptimized() {
for (int i = 0; i < selectedKeys.size; ++i) {
final SelectionKey k = selectedKeys.keys[i];
// 方便GC回收
selectedKeys.keys[i] = null;
final Object a = k.attachment();
// 根据类型不同执行不同的处理逻辑
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
// 一般不会走这个分支，除非用户主动注册NioTask到selector，netty单元测试里有案例
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
// 如果为true，则重置之后的所有selectKey，并调用selectNow()方法
// 因为run()方法执行本方法前已经置为false，所以不会进这里
if (needsToSelectAgain) {
selectedKeys.reset(i + 1);
selectAgain();
i = -1;
}
}
}
// 处理逻辑基本与processSelectedKeysOptimized相同
private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
if (selectedKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> i = selectedKeys.iterator();
for (;;) {
final SelectionKey k = i.next();
final Object a = k.attachment();
i.remove();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (!i.hasNext()) {
break;
}
if (needsToSelectAgain) {
selectAgain();
selectedKeys = selector.selectedKeys();
if (selectedKeys.isEmpty()) {
break;
} else {
i = selectedKeys.iterator();
}
}
}
}
既然内部逻辑类似，重点看一下processSelectedKeysOptimized()方法，NioTask分支一般不会走，感兴趣可以看一下netty的单元测试。重点看一下AbstractNioChannel分支，如果attachment是AbstractNioChannel类型，说明它是NioServerSocketChannel或者NioSocketChannel，需要进行IO读写相关的操作。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
// SelectionKey无效的处理
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
// channel没有关联的eventLoop直接返回
return;
}
// channel关联的eventLoop不是本eventLoop，直接返回，不应关闭channel
if (eventLoop != this || eventLoop == null) {
return;
}
// 关闭channel
unsafe.close(unsafe.voidPromise());
return;
}
try {
int readyOps = k.readyOps();
// 对于NioSocketChannel，连接需要先finishConnect才能继续读写
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// 下面3行的操作只是将OP_CONNECT从感兴趣选项中移除，防止重复触发
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// 说明有半包消息未发送完成，调用flush发送即可
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
// unsafe是多态，对于NioServerSocketChannel，read就是接受客户端TCP连接
// 对于NioSocketChannel，就是从channel中读取ByteBuffer
// 同时检测readyOps == 0 是解决JDK的循环bug
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
- runAllTasks()
这个是执行用户任务也就是非IO处理的方法，分为不指定时间和指定时间的两个重载方法。最大的不同就是带时间的方法是有执行时间限制的，防止用户任务长时间阻塞IO事件。
protected boolean runAllTasks() {
assert inEventLoop();
boolean fetchedAll;
boolean ranAtLeastOne = false;
do {
// 取一定时间段内的定时任务到普通任务队列里
fetchedAll = fetchFromScheduledTaskQueue();
// 运行任务队列里的任务
if (runAllTasksFrom(taskQueue)) {
ranAtLeastOne = true;
}
} while (!fetchedAll); // 取完所有定时任务为止
if (ranAtLeastOne) {
// 记录上次执行完任务的时间
lastExecutionTime = ScheduledFutureTask.nanoTime();
}
afterRunningAllTasks();
return ranAtLeastOne;
}
protected boolean runAllTasks(long timeoutNanos) {
fetchFromScheduledTaskQueue();
Runnable task = pollTask();
if (task == null) {
afterRunningAllTasks();
return false;
}
// 这是用户任务指定的截止时间
final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
long runTasks = 0;
long lastExecutionTime;
for (;;) {
safeExecute(task);
runTasks ++;
// nanoTime()是耗时的操作，所以这里每执行64个任务才检测一次是否超过时间
if ((runTasks & 0x3F) == 0) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
if (lastExecutionTime >= deadline) {
break;
}
}
// 执行任务
task = pollTask();
if (task == null) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
break;
}
}
afterRunningAllTasks();
this.lastExecutionTime = lastExecutionTime;
return true;
}
总结
从NioEventLoop的源码可以看到，netty在很多地方做了优化，还避免了很多JDK自带的bug
