Netty源码—3.Reactor线程模型二

大纲

5.NioEventLoop的执行总体框架

6.Reactor线程执行一次事件轮询

7.Reactor线程处理产生IO事件的Channel

8.Reactor线程处理任务队列之添加任务

9.Reactor线程处理任务队列之执行任务

10.NioEventLoop总结

 

5.NioEventLoop的执行总体框架

(1)Reactor线程所做的三件事情

(2)处理多久IO事件就执行多久任务

(3)NioEventLoop.run()方法的执行流程

 

(1)Reactor线程所做的三件事情

NioEventLoop的run()方法里有个无限for循环，for循环里便是Reactor线程所要做的3件事情。

 

一.首先是调用select()方法进行一次事件轮询

由于一个NioEventLoop对应一个Selector，所以该select()方法便是轮询注册到这个Reactor线程对应的Selector上的所有Channel的IO事件。注意，select()方法里也有一个无限for循环，但是这个无限for循环可能会被某些条件中断。

 

二.然后调用processSelectedKeys()方法处理轮询出来的IO事件

 

三.最后调用runAllTasks()方法来处理外部线程放入TaskQueue的任务

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {private volatile int ioRatio = 50;...@Overrideprotected void run() {for (;;) {...//1.调用select()方法执行一次事件轮询select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}...//2.处理产生IO事件的ChannelprocessSelectedKeys();...//3.执行外部线程放入TaskQueue的任务runAllTasks(ioTime * (100 - ioRatio) / ioRatio);}}private void select(boolean oldWakenUp) throws IOException {for(;;) {//1.定时任务截止时间快到了，中断本次轮询//2.轮询过程中发现有任务加入，中断本次轮询//3.阻塞式select操作: selector.select(timeoutMills)//4.避免JDK空轮询Bug}}...
}

(2)处理多久IO事件就执行多久任务

在NioEventLoop的run()方法中，有个ioRatio默认是50，代表处理IO事件的时间和执行任务的时间是1:1。也就是执行了多久的processSelectedKeys()方法后，紧接着就执行多久的runAllTasks()方法。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {private volatile int ioRatio = 50;...@Overrideprotected void run() {for (;;) {...//1.调用select()方法执行一次事件轮询select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}...final int ioRatio = this.ioRatio;if (ioRatio == 100) {try {processSelectedKeys();} finally {// Ensure we always run tasks.runAllTasks();}} else {final long ioStartTime = System.nanoTime();try {processSelectedKeys();} finally {// Ensure we always run tasks.final long ioTime = System.nanoTime() - ioStartTime;runAllTasks(ioTime * (100 - ioRatio) / ioRatio);}}...}}...
}

(3)NioEventLoop.run()方法的执行流程

NioEventLoop.run() -> for(;;)select() //执行一次事件轮询检查是否有IO事件processSelectedKeys() //处理产生IO事件的ChannelrunAllTasks() //处理异步任务队列
//这3步放在一个线程处理应该是为了节约线程，因为不是总会有IO事件和异步任务的

 

6.Reactor线程执行一次事件轮询

(1)执行select操作前设置wakeUp变量

(2)定时任务快开始了则中断本次轮询

(3)轮询中发现有任务加入则中断本次轮询

(4)执行阻塞式select操作

(5)避免JDK的空轮询Bug

(6)执行一次事件轮询的总结

 

(1)执行select操作前设置wakeUp变量

NioEventLoop有个wakenUp成员变量表示是否应该唤醒正在阻塞的select操作。NioEventLoop的run()方法准备执行select()方法进行一次新的循环逻辑之前，都会将wakenUp设置成false，标志新一轮循环的开始。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {//Boolean that controls determines if a blocked Selector.select should break out of its selection process. //In our case we use a timeout for the select method and the select method will block for that time unless waken up.private final AtomicBoolean wakenUp = new AtomicBoolean();...@Overrideprotected void run() {for (;;) {...//1.调用select()方法执行一次事件轮询select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}...}}...
}

如下是NioEventLoop的select()方法的执行逻辑，也就是Netty关于事件循环的4段逻辑。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;...private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;for(;;) {//1.定时任务截止时间快到了，中断本次轮询//2.轮询过程中发现有任务加入，中断本次轮询//3.阻塞式select操作: selector.select(timeoutMills)//4.避免JDK空轮询Bug}}...
}

(2)定时任务快开始了则中断本次轮询

NioEventLoop中的Reactor线程的select操作也是一个for循环。

 

在for循环第一步，如果发现当前定时任务队列中某个任务的开始时间快到了(小于0.5ms)，那么就跳出循环。在跳出循环之前，如果发现目前为止还没有进行过select操作，就调用一次selectNow()方法执行非阻塞式select操作。

 

Netty里的定时任务队列是按照延迟时间从小到大进行排序的，所以delayNanos()方法返回的第一个定时任务的延迟时间便是最早截止的时间。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;...private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;int selectCnt = 0;long currentTimeNanos = System.nanoTime();//当前时间long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);//当前时间 + 定时任务的最早截止时间for(;;) {//1.定时任务截止时间快到了，中断本次轮询long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;if (timeoutMillis <= 0) {if (selectCnt == 0) {selector.selectNow();//非阻塞执行select操作selectCnt = 1;}break;}...}}...
}//Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread.
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {...protected long delayNanos(long currentTimeNanos) {ScheduledFutureTask<?> scheduledTask = peekScheduledTask();if (scheduledTask == null) {return SCHEDULE_PURGE_INTERVAL;}return scheduledTask.delayNanos(currentTimeNanos);}...
}//Abstract base class for EventExecutors that want to support scheduling.
public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {Queue<ScheduledFutureTask<?>> scheduledTaskQueue;//定时任务队列...final ScheduledFutureTask<?> peekScheduledTask() {Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;if (scheduledTaskQueue == null) {return null;}return scheduledTaskQueue.peek();}...
}final class ScheduledFutureTask<V> extends PromiseTask<V> implements ScheduledFuture<V> {...public long delayNanos(long currentTimeNanos) {return Math.max(0, deadlineNanos() - (currentTimeNanos - START_TIME));}public long deadlineNanos() {return deadlineNanos;}...
}

(3)轮询中发现有任务加入则中断本次轮询

注意：Netty的任务队列包括普通任务和定时任务。定时任务快开始时需要中断本次轮询，普通任务队列非空时也需要中断本次轮询。

 

Netty为了保证普通任务队列里的普通任务能够及时执行，在调用selector.select()方法进行阻塞式select操作前会判断普通任务队列是否为空。如果不为空，那么就调用selector.selectNow()方法执行一次非阻塞select操作，然后跳出循环。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;...private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;int selectCnt = 0;...for(;;) {...//2.轮询过程中发现有任务加入，中断本次轮询if (hasTasks() && wakenUp.compareAndSet(false, true)) {selector.selectNow();//非阻塞式执行select操作selectCnt = 1;break;}...}}...
}//Abstract base class for EventLoops that execute all its submitted tasks in a single thread.
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {private final Queue<Runnable> tailTasks;...@Overrideprotected boolean hasTasks() {return super.hasTasks() || !tailTasks.isEmpty();}...
}//Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread.
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {private final Queue<Runnable> taskQueue;//普通任务队列...protected boolean hasTasks() {assert inEventLoop();return !taskQueue.isEmpty();}...
}

(4)执行阻塞式select操作

一.最多阻塞到第一个定时任务的开始时间

二.外部线程提交任务会唤醒Reactor线程

三.是否中断本次轮询的判断条件

 

一.最多阻塞到第一个定时任务的开始时间

执行到这一步，说明Netty的普通任务队列里的队列为空，并且所有定时任务的开始时间还未到(大于0.5ms)。于是便进行一次阻塞式select操作，一直阻塞到第一个定时任务的开始时间，也就是把timeoutMills作为参数传入select()方法中。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;...private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;...for(;;) {...//3.阻塞式select操作: selector.select(timeoutMills)，最多阻塞timeoutMills时间int selectedKeys = selector.select(timeoutMillis);selectCnt ++;if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {// - Selected something,// - waken up by user, or// - the task queue has a pending task.// - a scheduled task is ready for processingbreak;}...}}...
}

二.外部线程提交任务会唤醒Reactor线程

如果第一个定时任务的延迟时间非常长，比如一小时，那么有可能线程会一直阻塞在select操作(select完还是会返回的)。但只要这段时间内有新任务加入，该阻塞就会被释放。

 

比如当有外部线程执行NioEventLoop的execute()方法添加任务时，就会调用NioEventLoop的wakeUp()方法来通过selector.wakeup()方法，去唤醒正在执行selector.select(timeoutMills)而被阻塞的线程。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {//Boolean that controls determines if a blocked Selector.select should break out of its selection process. //In our case we use a timeout for the select method and the select method will block for that time unless waken up.private final AtomicBoolean wakenUp = new AtomicBoolean();Selector selector;...@Overrideprotected void wakeup(boolean inEventLoop) {if (!inEventLoop && wakenUp.compareAndSet(false, true)) {selector.wakeup();}}
}//Abstract base class for EventLoops that execute all its submitted tasks in a single thread.
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {...
}//Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread.
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {...@Overridepublic void execute(Runnable task) {if (task == null) {throw new NullPointerException("task");}boolean inEventLoop = inEventLoop();if (inEventLoop) {addTask(task);} else {startThread();addTask(task);if (isShutdown() && removeTask(task)) {reject();}}if (!addTaskWakesUp && wakesUpForTask(task)) {//调用NioEventLoop.wakeup()方法唤醒正在执行selector.select(timeoutMills)而被阻塞的线程wakeup(inEventLoop);}}...
}

三.是否中断本次轮询的判断条件

阻塞式select操作结束后，Netty又会做一系列状态判断来决定是否中断本次轮询，如果满足如下条件就中断本次轮询：

条件一：检测到IO事件

条件二：被用户主动唤醒

条件三：普通任务队列里有任务需要执行

条件四：第一个定时任务即将要被执行

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;...private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;...for(;;) {...//3.阻塞式select操作: selector.select(timeoutMills)，最多阻塞timeoutMills时间int selectedKeys = selector.select(timeoutMillis);selectCnt ++;//阻塞式select操作结束后，Netty又会做一系列状态判断来决定是否中断本次轮询if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {// - Selected something, 检测到IO事件// - waken up by user, 被用户主动唤醒// - the task queue has a pending task. 普通任务队列里有任务需要执行// - a scheduled task is ready for processing 第一个定时任务即将要被执行break;}...}}...
}

(5)避免JDK的空轮询Bug

JDK空轮询Bug会导致selector一直空轮询，最终导致CPU的利用率100%。

 

一.Netty避免JDK空轮询的方法

首先每次执行selector.select(timeoutMillis)之前都会记录开始时间，在阻塞式select操作后记录结束时间。

 

然后判断阻塞式select操作是否持续了至少timeoutMillis时间。如果阻塞式select操作持续的时间大于等于timeoutMillis，说明这是一次有效的轮询，于是重置selectCnt为1。如果阻塞式select操作持续的时间小于timeoutMillis，则说明可能触发了JDK的空轮询Bug，于是自增selectCnt。当持续时间很短的select操作的次数selectCnt超过了512次，那么就重建Selector。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;private static final int SELECTOR_AUTO_REBUILD_THRESHOLD = 512;...private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;intselectCnt = 0;long currentTimeNanos = System.nanoTime();//记录开始时间...for(;;) {...int selectedKeys = selector.select(timeoutMillis);//进行阻塞式select操作selectCnt++;//select操作持续时间很短，可能出现空轮询，selectCnt需要自增long time = System.nanoTime();//记录结束时间if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {//如果select操作持续的时间大于timeoutMillis，说明这是一次有效的轮询，重置selectCnt为1selectCnt = 1;} else if (selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {//如果持续时间很短的select操作的次数超过了512次，就重建selectorrebuildSelector();//重建Selectorselector = this.selector;selector.selectNow();selectCnt = 1;break;}currentTimeNanos = time;...}}...
}

二.重建Selector的逻辑

重建Selector的逻辑就是通过openSelector()方法创建一个新的Selector，然后执行一个无限的for循环，只要执行过程中出现一次并发修改SelectionKeys异常，那么就重新开始转移，直到转移完成。

 

具体的转移步骤为：首先拿到有效的key，然后取消该key在旧Selector上的事件注册。接着将该key对应的Channel注册到新的Selector上，最后重新绑定Channel和新的key。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;...//Replaces the current Selector of this event loop with newly created Selectors to work around the infamous epoll 100% CPU bug.public void rebuildSelector() {final Selector oldSelector = selector;final Selector newSelector = openSelector();int nChannels = 0;for (;;) {try {for (SelectionKey key: oldSelector.keys()) {Object a = key.attachment();//1.拿到有效的keyif (!key.isValid() || key.channel().keyFor(newSelector) != null) {continue;}int interestOps = key.interestOps();//2.取消该key在旧Selector上的事件注册key.cancel();//3.将该key对应的Channel注册到新的Selector上SelectionKey newKey = key.channel().register(newSelector, interestOps, a);if (a instanceof AbstractNioChannel) {//4.重新绑定Channel和新的key((AbstractNioChannel) a).selectionKey = newKey;}nChannels++;}break;} catch(ConcurrentModificationException e) {continue;}}selector = newSelector;oldSelector.close();}...
}

(6)执行一次事件轮询的总结

关于Reactor线程的select操作所做的事情：

 

简单来说就是：

不断轮询是否有IO事件发生，并且在轮询过程中不断检查是否有任务需要执行，从而保证Netty任务队列中的任务都能够及时执行，以及在轮询过程中会巧妙地使用一个计数器来避开JDK的空轮询Bug。

 

详细来说就是：

NioEventLoop的select()方法首先会判断有没有定时任务快到要开始的时间了、普通任务队列taskQueue里是否存在任务。如果有就调用selector.selectNow()进行非阻塞式的select操作，如果都没有就调用selector.select(timeoutMillis)进行阻塞式select操作。在阻塞式select操作结束后，会判断这次select操作是否阻塞了timeoutMillis这么长时间。如果没有阻塞那么长时间就表明可能触发了JDK的空轮询Bug，接下来就会继续判断可能触发空轮询Bug的次数是否达到了512次，如果达到了就通过替换原来Selector的方式去避开空轮询Bug。

 

7.Reactor线程处理产生IO事件的Channel

(1)处理IO事件的关键逻辑

(2)Netty对selectedKeys的优化

(3)处理IO事件的过程说明

(4)处理IO事件的总结

 

(1)处理IO事件的关键逻辑

Reactor线程执行的第一步是轮询出注册在Selector上的IO事件，第二步便是处理这些IO事件了。

 

processSelectedKeys()的关键逻辑包含两部分：

一.针对selectedKeys的优化

二.processSelectedKeysOptimized()方法真正处理IO事件

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;private SelectedSelectionKeySet selectedKeys;...@Overrideprotected void run() {for (;;) {...//1.调用select()方法执行一次事件轮询select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}...//2.处理产生IO事件的ChannelprocessSelectedKeys();...//3.执行外部线程放入TaskQueue的任务runAllTasks(ioTime * (100 - ioRatio) / ioRatio);}}private void processSelectedKeys() {if (selectedKeys != null) {//selectedKeys.flip()会返回一个数组processSelectedKeysOptimized(selectedKeys.flip());} else {processSelectedKeysPlain(selector.selectedKeys());}}...
}

(2)Netty对selectedKeys的优化

Netty对selectedKeys的所有优化都是在NioEventLoop的openSelector()方法中体现的。

 

这个优化指的是：

Selector.select()操作每次都会把就绪状态的IO事件添加到Selector底层的两个HashSet成员变量中，而Netty会通过反射的方式将Selector中用于存放SelectionKey的HashSet替换成数组，使得添加SelectionKey的时间复杂度由HashSet的O(n)降为数组的O(1)。

 

具体来说就是：

NioEventLoop的成员变量selectedKeys是一个SelectedSelectionKeySet对象，会在NioEventLoop的openSelector()方法中创建。之后openSelector()方法会通过反射将selectedKeys与Selector的两个成员变量绑定。SelectedSelectionKeySet继承了AbstractSet，但底层是使用数组来存放SelectionKey的。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;private SelectedSelectionKeySet selectedKeys;...private Selector openSelector() {final Selector selector = provider.openSelector();final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();...//下面的selectorImplClass对应于sun.nio.ch.SelectorImplField selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");selectedKeysField.setAccessible(true);publicSelectedKeysField.setAccessible(true);//这两个set才是优化中的精华，一句话总结就是：//用数组来替换Selector中用于存放SelectionKey的HashSet的实现，做到add()方法的时间复杂度为O(1)selectedKeysField.set(selector, selectedKeySet);publicSelectedKeysField.set(selector, selectedKeySet);...selectedKeys = selectedKeySet;...}...
}//SelectedSelectionKeySet继承了AbstractSet，说明该类可以当作一个Set来用，但是底层使用两个数组来交替使用
//在add()方法中，首先判断当前应该使用哪个数组，然后找到对应的数组执行如下3个步骤：
//步骤一：将SelectionKey放入数组的尾部
//步骤二：更新该数组的逻辑长度+1
//步骤三：如果该数组的逻辑长度等于该数组的物理长度，就将该数组扩容//待程序运行一段时间后，等数组的长度足够长，每次轮询到NIO事件的时候，
//调用这里的add()方法只需要O(1)的时间复杂度就能将SelectionKey放入到Set中，
//而JDK底层使用的HashSet的put()方法的时间复杂度最小是O(1)、最大是O(n)，
//使用数组替换HashSet还有一个好处就是遍历的时候非常高效
final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {private SelectionKey[] keysA;private int keysASize;private SelectionKey[] keysB;private int keysBSize;private boolean isA = true;SelectedSelectionKeySet() {keysA = new SelectionKey[1024];keysB = keysA.clone();}@Overridepublic boolean add(SelectionKey o) {if (o == null) {return false;}if (isA) {int size = keysASize;keysA[size ++] = o;keysASize = size;if (size == keysA.length) {doubleCapacityA();}} else {int size = keysBSize;keysB[size ++] = o;keysBSize = size;if (size == keysB.length) {doubleCapacityB();}}return true;}//返回一个数组SelectionKey[] flip() {if (isA) {isA = false;keysA[keysASize] = null;keysBSize = 0;return keysA;} else {isA = true;keysB[keysBSize] = null;keysASize = 0;return keysB;}}...
}//可以看到，SelectorImpl的两个成员变量selectedKeys和keys都是HashSet
public abstract class SelectorImpl extends AbstractSelector {protected Set<SelectionKey> selectedKeys = new HashSet();protected HashSet<SelectionKey> keys = new HashSet();private Set<SelectionKey> publicKeys;private Set<SelectionKey> publicSelectedKeys;protected SelectorImpl(SelectorProvider var1) {super(var1);if (Util.atBugLevel("1.4")) {this.publicKeys = this.keys;this.publicSelectedKeys = this.selectedKeys;} else {this.publicKeys = Collections.unmodifiableSet(this.keys);this.publicSelectedKeys = Util.ungrowableSet(this.selectedKeys);}}...
}

(3)处理IO事件的过程说明

说明一：

首先取出IO事件。IO事件是以数组的形式从selectedKeys中取的，其对应的Channel则由SelectionKey的attachment()方法返回。

 

此时可以体会到优化过的selectedKeys的好处。因为遍历时遍历的是数组，相对JDK原生的HashSet，效率有所提高。

 

拿到当前的SelectionKey之后，便将selectedKeys[i]设置为null，这样做是为了方便GC。因为假设一个NioEventLoop平均每次轮询出N个IO事件，高峰期轮询出3N个事件，那么selectedKeys的物理长度要大于等于3N。如果每次处理这些key时不设置selectedKeys[i]为null，那么高峰期一过，这些保存在数组尾部的selectedKeys[i]对应的SelectionKey将一直无法被回收，虽然SelectionKey对应的对象可能不大，但其关联的attachment则可能很大。这些对象如果一直存活无法回收，就可能发生内存泄露。

 

说明二：

然后获取当前SelectionKey对应的attachment。这个attachement就是取出的IO事件对应的Channel了，于是接下来就可以处理该Channel了。

 

由于Netty在注册服务端Channel时，会将AbstractNioChannel内部的SelectableChannel对象注册到Selector对象上，并且将AbstractNioChannel作为SelectableChannel对象的一个attachment附属。所以当JDK轮询出某个SelectableChannel有IO事件时，就可以通过attachment()方法直接取出AbstractNioChannel进行操作了。

 

说明三：

接着便会调用processSelectedKey()方法对SelectionKey和AbstractNioChannel进行处理。Netty有两大类Channel：一个是NioServerSocketChannel，由bossGroup处理。另一个是NioSocketChannel，由workerGroup处理。对于boss的NioEventLoop来说，轮询到的是连接事件。对于worker的NioEventLoop来说，轮询到的是读写事件。

 

说明四：

最后会判断是否再进行一次轮询。NioEventLoop的run()方法每次在轮询到IO事件后，都会将needsToSelectAgain设置为false。只有当Channel从Selector上移除时，也就是调用NioEventLoop的cancel()方法时，发现被取消的key已经达到256次了，才会将needsToSelectAgain设置为true。当needsToSelectAgain为true，就会调用selectAgain()方法再进行一次轮询。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop.
public final class NioEventLoop extends SingleThreadEventLoop {Selector selector;private SelectedSelectionKeySet selectedKeys;private boolean needsToSelectAgain;private int cancelledKeys;private static final int CLEANUP_INTERVAL = 256;...@Overrideprotected void run() {for (;;) {...//1.调用select()方法执行一次事件轮询select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}...//2.处理产生IO事件的ChannelneedsToSelectAgain = false;processSelectedKeys();...//3.执行外部线程放入TaskQueue的任务runAllTasks(ioTime * (100 - ioRatio) / ioRatio);}}private void processSelectedKeys() {if (selectedKeys != null) {//selectedKeys.flip()会返回一个数组processSelectedKeysOptimized(selectedKeys.flip());} else {processSelectedKeysPlain(selector.selectedKeys());}}private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {for (int i = 0;; i ++) {//1.首先取出IO事件final SelectionKey k = selectedKeys[i];if (k == null) {break;}selectedKeys[i] = null;//Help GC//2.然后获取对应的Channel和处理该Channel//默认情况下，这个a就是NioChannel，也就是服务端启动时经过Netty封装的Channelfinal Object a = k.attachment();if (a instanceof AbstractNioChannel) {//网络事件的处理processSelectedKey(k, (AbstractNioChannel) a);} else {//NioTask主要用于当一个SelectableChannel注册到Selector时，执行的一些任务NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;processSelectedKey(k, task);}//3.最后判断是否应该再进行一次轮询if (needsToSelectAgain) {for (;;) {i++;if (selectedKeys[i] == null) {break;}selectedKeys[i] = null;}selectAgain();//selectedKeys.flip()会返回一个数组selectedKeys = this.selectedKeys.flip();i = -1;}}}private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();if (!k.isValid()) {final EventLoop eventLoop;try {eventLoop = ch.eventLoop();} catch (Throwable ignored) {//If the channel implementation throws an exception because there is no event loop, //we ignore this because we are only trying to determine if ch is registered to this event loop and thus has authority to close ch.return;}//Only close ch if ch is still registerd to this EventLoop. //ch could have deregistered from the event loop and thus the SelectionKey could be cancelled as part of the deregistration process, //but the channel is still healthy and should not be closed.if (eventLoop != this || eventLoop == null) {return;}//close the channel if the key is not valid anymoreunsafe.close(unsafe.voidPromise());return;}try {int readyOps = k.readyOps();//We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise//the NIO JDK channel implementation may throw a NotYetConnectedException.if ((readyOps & SelectionKey.OP_CONNECT) != 0) {//remove OP_CONNECT as otherwise Selector.select(..) will always return without blockingint ops = k.interestOps();ops &= ~SelectionKey.OP_CONNECT;k.interestOps(ops);unsafe.finishConnect();}//Process OP_WRITE first as we may be able to write some queued buffers and so free memory.if ((readyOps & SelectionKey.OP_WRITE) != 0) {//Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to writech.unsafe().forceFlush();}//Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead to a spin loopif ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {unsafe.read();if (!ch.isOpen()) {//Connection already closed - no need to handle write.return;}}} catch (CancelledKeyException ignored) {unsafe.close(unsafe.voidPromise());}}void cancel(SelectionKey key) {key.cancel();cancelledKeys ++;if (cancelledKeys >= CLEANUP_INTERVAL) {cancelledKeys = 0;needsToSelectAgain = true;}}...
}

(4)处理IO事件的总结

Netty默认情况下会通过反射，将Selector底层用于存放SelectionKey的两个HashSet，转化成一个数组来提升处理IO事件的效率。

 

在处理每一个SelectionKey时都会拿到对应的一个attachment，而这个attachment就是在服务端Channel注册Selector时所绑定的一个AbstractNioChannel。所以在处理每一个SelectionKey时，都可以找到对应的AbstractNioChannel，然后通过Pipeline将处理串行到ChannelHandler，回调到用户的方法。

 

8.Reactor线程处理任务队列之添加任务

(1)Reactor线程执行一次事件轮询的过程

(2)任务的分类和添加说明

(3)普通任务的添加

(4)定时任务的添加

(5)Netty的定时任务机制补充

 

(1)Reactor线程执行一次事件轮询的过程

Reactor线程通过NioEventLoop的run()方法每进行一次事件轮询，首先会调用select()方法尝试检测出IO事件，然后会调用processSelectedKeys()方法处理检测出的IO事件。其中IO事件主要包括新连接接入事件和连接的数据读写事件，最后会调用runAllTasks()方法处理任务队列中的异步任务。

 

(2)任务的分类和添加说明

runAllTasks()方法中的Task包括普通任务和定时任务，分别存放于NioEventLoop不同的队列里。一个是普通的任务队列MpscQueue，另一个是定时的任务队列PriorityQueue。

 

普通的任务队列MpscQueue在创建NioEventLoop时创建的，然后在外部线程调用NioEventLoop的execute()方法时，会调用addTask()方法将Task保存到普通的任务队列里。

 

定时的任务队列PriorityQueue则是在添加定时任务时创建的，然后在外部线程调用NioEventLoop的schedule()方法时，会调用scheduleTaskQueue().add()方法将Task保存到定时的任务队列里。

 

(3)普通任务的添加

场景一：用户自定义普通任务

场景二：外部线程调用Channel的方法

 

当通过ctx.channel().eventLoop().execute(...)自定义普通任务，或者通过非Reactor线程(外部线程)调用Channel的各类方法时，最后都会执行到SingleThreadEventExecutor的execute()方法。

 

场景一：用户自定义普通任务

不管是外部线程还是Reactor线程执行NioEventLoop的execute()方法，都会调用NioEventLoop的addTask()方法，然后调用offerTask()方法。而offerTask()方法会使用一个taskQueue将Task保存起来。这个taskQueue其实就是一个MPSC队列，每一个NioEventLoop都会有一个MPSC队列。

 

Netty使用MPSC队列可以方便地将外部线程的异步任务进行聚集，然后在Reactor线程内部用单线程来批量执行以提升性能。可以借鉴Netty的这种任务执行模式来处理类似多线程数据聚合，定时上报应用。

//场景一：用户自定义普通任务
ctx.channel().eventLoop().execute(new Runnable() {@Overridepublic void run() {...    }
});public final class NioEventLoop extends SingleThreadEventLoop {...
}public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {...
}public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {//每一个NioEventLoop会有一个MPSC队列private final Queue<Runnable> taskQueue;protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {super(parent);this.addTaskWakesUp = addTaskWakesUp;this.maxPendingTasks = Math.max(16, maxPendingTasks);this.executor = ObjectUtil.checkNotNull(executor, "executor");//创建普通的任务队列MpscQueuetaskQueue = newTaskQueue(this.maxPendingTasks);rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");}...@Overridepublic void execute(Runnable task) {if (task == null) throw new NullPointerException("task");boolean inEventLoop = inEventLoop();//不管是外部线程还是Reactor线程执行NioEventLoop的execute()方法，都会调用NioEventLoop的addTask()方法if (inEventLoop) {addTask(task);} else {startThread();addTask(task);if (isShutdown() && removeTask(task)) {reject();}}if (!addTaskWakesUp && wakesUpForTask(task)) wakeup(inEventLoop);}//Add a task to the task queue, or throws a RejectedExecutionException if this instance was shutdown before.protected void addTask(Runnable task) {if (task == null) throw new NullPointerException("task");if (!offerTask(task)) reject(task);}final boolean offerTask(Runnable task) {if (isShutdown()) reject();return taskQueue.offer(task);}...
}

场景二：外部线程调用Channel的方法

这个场景是在业务线程里，根据用户标识找到对应的Channel，然后调用Channel的write()方法向该用户推送消息。

 

外部线程在调用Channel的write()方法时，executor.inEventLoop()会返回false。于是会将write操作封装成一个WriteTask，然后调用safeExecute()方法来执行。默认情况下会获取Channel对应的NIO线程，然后作为参数传入safeExecute()方法中进行执行，从而确保任务会由Channel对应的NIO线程执行，通过单线程执行来实现线程安全。

//场景二：当前线程为业务线程
channel.write(...)abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext, ResourceLeakHint {final EventExecutor executor;//一般初始化时默认为nullAbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name, Class<? extends ChannelHandler> handlerClass) {this.pipeline = pipeline;this.executor = executor;...}...private void write(Object msg, boolean flush, ChannelPromise promise) {AbstractChannelHandlerContext next = findContextOutbound();final Object m = pipeline.touch(msg, next);EventExecutor executor = next.executor();//外部线程在调用Channel的write()方法时，executor.inEventLoop()会返回falseif (executor.inEventLoop()) {if (flush) {next.invokeWriteAndFlush(m, promise);} else {next.invokeWrite(m, promise);}} else {//将write操作封装成一个WriteTaskAbstractWriteTask task;if (flush) {task = WriteAndFlushTask.newInstance(next, m, promise);}  else {task = WriteTask.newInstance(next, m, promise);}//调用safeExecute()来执行safeExecute(executor, task, promise, m);}}@Overridepublic EventExecutor executor() {if (executor == null) {return channel().eventLoop();} else {return executor;}}private AbstractChannelHandlerContext findContextOutbound(int mask) {AbstractChannelHandlerContext ctx = this;EventExecutor currentExecutor = executor();do {ctx = ctx.prev;} while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_OUTBOUND));return ctx;}private static void safeExecute(EventExecutor executor, Runnable runnable, ChannelPromise promise, Object msg) {try {//调用SingleThreadEventExecutor.execute()方法executor.execute(runnable);} catch (Throwable cause) {try {promise.setFailure(cause);} finally {if (msg != null) {ReferenceCountUtil.release(msg);}}}}...
}public class DefaultChannelPipeline implements ChannelPipeline {...final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {private final Unsafe unsafe;HeadContext(DefaultChannelPipeline pipeline) {//传入的executor为nullsuper(pipeline, null, HEAD_NAME, HeadContext.class);unsafe = pipeline.channel().unsafe();setAddComplete();}...}final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {TailContext(DefaultChannelPipeline pipeline) {//传入的executor为nullsuper(pipeline, null, TAIL_NAME, TailContext.class);setAddComplete();}...}...
}public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {//每一个NioEventLoop会有一个MPSC队列private final Queue<Runnable> taskQueue;protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {super(parent);this.addTaskWakesUp = addTaskWakesUp;this.maxPendingTasks = Math.max(16, maxPendingTasks);this.executor = ObjectUtil.checkNotNull(executor, "executor");//创建普通的任务队列MpscQueuetaskQueue = newTaskQueue(this.maxPendingTasks);rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");}...@Overridepublic void execute(Runnable task) {if (task == null) throw new NullPointerException("task");boolean inEventLoop = inEventLoop();//不管是外部线程还是Reactor线程执行NioEventLoop的execute()方法，都会调用NioEventLoop的addTask()方法if (inEventLoop) {addTask(task);} else {startThread();addTask(task);if (isShutdown() && removeTask(task)) {reject();}}if (!addTaskWakesUp && wakesUpForTask(task)) wakeup(inEventLoop);}//Add a task to the task queue, or throws a RejectedExecutionException if this instance was shutdown before.protected void addTask(Runnable task) {if (task == null) throw new NullPointerException("task");if (!offerTask(task)) reject(task);}final boolean offerTask(Runnable task) {if (isShutdown()) reject();return taskQueue.offer(task);}...
}

(4)定时任务的添加

通常使用ctx.channel().eventLoop().schedule(..)自定义定时任务，其中schedule()方法会通过scheduledTaskQueue().add(task)来添加定时任务。首先scheduledTaskQueue()方法会返回一个优先级队列，然后通过该优先级队列的add()方法将定时任务对象加入到队列中。

 

注意，这里可以直接使用优先级队列而不用考虑多线程并发问题的原因如下。如果是外部线程调用schedule()方法添加定时任务，那么Netty会将添加定时任务这个逻辑封装成一个普通的Task。这个Task的任务是一个"添加某定时任务"的任务，而不是添加某定时任务。这样，对优先级队列的访问就变成单线程了，也就是只有Reactor线程会访问，从而不存在多线程并发问题。

//场景三：用户自定义定时任务，这也是用得最多的方法
ctx.channel().eventLoop().schedule(new Runnable() {@Overridepublic void run() {...    }
}, 60, TimeUnit.SECONDS);//Abstract base class for EventExecutors that want to support scheduling.
public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {Queue<ScheduledFutureTask<?>> scheduledTaskQueue;...//添加定时任务<V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {if (inEventLoop()) {//如果当前线程是Reactor线程，则直接往PriorityQueue中添加任务scheduledTaskQueue().add(task);} else {//如果是外部线程，则调用SingleThreadEventExecutor.execute()方法//将添加定时任务这一动作也封装成一个普通任务execute(new Runnable() {@Overridepublic void run() {scheduledTaskQueue().add(task);}});}return task;}Queue<ScheduledFutureTask<?>> scheduledTaskQueue() {if (scheduledTaskQueue == null) {//创建定时的任务队列PriorityQueuescheduledTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>();}return scheduledTaskQueue;}...
}

(5)Netty的定时任务机制补充

一.如何保证截止时间最近的任务优先执行

为什么定时任务要保存在优先级队列中？优先级队列的特性是会按照一定的顺序来排列内部元素，内部元素是可以比较的。由于优先级队列中的每个元素都是定时任务，所以定时任务也是可以比较的。比较的逻辑就是：先比较定时任务的截止时间，在截止时间相同的情况下再比较定时任务的添加顺序也就是ID。

 

二.Netty的定时任务有三种执行方式

方式一：定时任务不会被重复执行

ctx.channel().eventLoop().schedule()，传递的periodNanos为0。

方式二：每隔一段时间执行一次

ctx.channel().eventLoop().scheduleAtFixedRate()，传递的periodNanos为正数。

方式三：隔相同时间再执行一次

ctx.channel().eventLoop().scheduleWithFixedDelay()，传递的periodNanos为负数。

 

Netty的3种定时任务的执行逻辑是通过调整下一次任务的截止时间来运行的。首先修改完下一次执行的截止时间，然后把当前任务再次加入队列，这样就能确保任务在适当的时候执行。

final class ScheduledFutureTask<V> extends PromiseTask<V> implements ScheduledFuture<V> {//每个定时任务都有一个唯一的IDprivate static final AtomicLong nextTaskId = new AtomicLong();private final long id = nextTaskId.getAndIncrement();private long deadlineNanos;//标识一个任务是否重复执行，以及以何种方式执行private final long periodNanos;...@Overridepublic int compareTo(Delayed o) {if (this == o) {return 0;}ScheduledFutureTask<?> that = (ScheduledFutureTask<?>) o;long d = deadlineNanos() - that.deadlineNanos();if (d < 0) {return -1;} else if (d > 0) {return 1;} else if (id < that.id) {return -1;} else if (id == that.id) {throw new Error();} else {return 1;}}public long deadlineNanos() {return deadlineNanos;}@Overridepublic void run() {...if (periodNanos == 0) {//1.对应schedule()方法，表示一次性任务V result = task.call();setSuccessInternal(result);} else {task.call();long p = periodNanos;if (p > 0) {//2.对应scheduleAtFixedRate()方法，表示以固定速率执行任务deadlineNanos += p;} else {//3.对应scheduleWithFixedDelay()方法，表示以固定的延时执行任务deadlineNanos = nanoTime() - p;}scheduledTaskQueue.add(this);}...}...
}

 

9.Reactor线程处理任务队列之执行任务

(1)runAllTasks()方法需要传入超时时间

(2)Reactor线程执行任务的步骤

(3)Netty性能优化之批量策略

(4)NioEventLoop.run()方法执行任务总结

 

(1)runAllTasks()方法需要传入超时时间

SingleThreadEventExecutor的runAllTasks()方法需要传入参数timeoutNanos，表示尽量在timeoutNanos时间内将所有的任务都取出来执行一遍。因为如果Reactor线程在执行任务时停留的时间过长，那么将会累积许多IO事件无法及时处理，从而导致大量客户端请求阻塞。因此Netty会精细控制内部任务队列的执行时间。

 

(2)Reactor线程执行任务的步骤

一.任务聚合

转移定时任务到MPSC队列，这里只是将快到期的定时任务转移到MPSC队列里。

 

二.时间计算

计算本轮任务执行的截止时间，此时所有截止时间已到达的定时任务均被填充到普通的任务队列(MPSC队列)里了。

 

三.任务执行

首先不抛异常地同步执行任务，然后累加当前已执行的任务数，接着每隔64次计算一下当前时间是否已超截止时间，最后判断本轮任务是否已经执行完毕。

//Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread.
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {//每一个NioEventLoop会有一个MPSC队列private final Queue<Runnable> taskQueue;...//Poll all tasks from the task queue and run them via Runnable#run() method.//This method stops running the tasks in the task queue and returns if it ran longer than timeoutNanos.protected boolean runAllTasks(long timeoutNanos) {//1.转移定时任务到MPSC队列，也就是任务聚合fetchFromScheduledTaskQueue();//从普通的任务队列(MPSC队列)里获取任务Runnable task = pollTask();if (task == null) {afterRunningAllTasks();return false;}//2.计算本轮任务执行的截止时间final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;long runTasks = 0;long lastExecutionTime;//3.执行任务，通过for循环逐个执行pollTask()取出的任务for (;;) {//3.1 不抛异常地执行任务(同步阻塞)，确保任务可以安全执行safeExecute(task);//3.2 累加当前已执行的任务数runTasks ++;//3.3 每隔64次计算一下当前时间是否已经超过截止时间，因为ScheduledFutureTask.nanoTime()也挺耗时的if ((runTasks & 0x3F) == 0) {lastExecutionTime = ScheduledFutureTask.nanoTime();if (lastExecutionTime >= deadline) {break;}}//3.4 判断本轮任务是否已经执行完毕task = pollTask();if (task == null) {lastExecutionTime = ScheduledFutureTask.nanoTime();break;}}afterRunningAllTasks();this.lastExecutionTime = lastExecutionTime;return true;}private boolean fetchFromScheduledTaskQueue() {long nanoTime = AbstractScheduledEventExecutor.nanoTime();Runnable scheduledTask  = pollScheduledTask(nanoTime);while (scheduledTask != null) {if (!taskQueue.offer(scheduledTask)) {scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);return false;}scheduledTask  = pollScheduledTask(nanoTime);}return true;}protected Runnable pollTask() {assert inEventLoop();return pollTaskFrom(taskQueue);}protected final Runnable pollTaskFrom(Queue<Runnable> taskQueue) {for (;;) {Runnable task = taskQueue.poll();if (task == WAKEUP_TASK) {continue;}return task;}}...
}//Abstract base class for EventExecutors that want to support scheduling.
public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {Queue<ScheduledFutureTask<?>> scheduledTaskQueue;...//Return the Runnable which is ready to be executed with the given nanoTime.//You should use #nanoTime() to retrieve the the correct nanoTime.protected final Runnable pollScheduledTask(long nanoTime) {assert inEventLoop();Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();if (scheduledTask == null) {return null;}if (scheduledTask.deadlineNanos() <= nanoTime) {scheduledTaskQueue.remove();return scheduledTask;}return null;}...
}//Abstract base class for EventExecutor implementations.
public abstract class AbstractEventExecutor extends AbstractExecutorService implements EventExecutor {...//Try to execute the given Runnable and just log if it throws a Throwable.protected static void safeExecute(Runnable task) {try {task.run();//同步执行任务} catch (Throwable t) {logger.warn("A task raised an exception. Task: {}", task, t);}}...
}

(3)Netty性能优化之间隔策略

假设任务队列里有海量的小任务，如果每次执行完任务都需要判断是否到截止时间，那么效率是比较低的。所以Netty选择通过每隔64个任务才判断一下是否到截止时间，那么效率就会高很多。

 

(4)NioEventLoop.run()方法执行任务总结

Netty里的任务分两种：一种是普通的任务，另一种是定时的任务。Netty在执行这些任务时首先会把定时任务聚合到普通任务队列里，然后再从普通任务队列里获取任务逐个执行，并且是每执行64个任务之后才判断一下当前时间是否超过最大允许执行时间。如果超过就直接中断，中断之后就会进行下一次NioEventLoop.run()方法的for循环。

 

10.NioEventLoop总结

(1)NioEventLoop的执行流程总结

(2)Reactor线程模型总结

(3)NioEventLoop创建启动执行的总结

 

(1)NioEventLoop的执行流程总结

一.NioEventLoop在执行过程中首先会不断检测是否有IO事件发生，然后如果检测出有IO事件就处理IO事件，接着处理完IO事件之后再处理外部线程提交过来的异步任务。

 

二.在检测是否有IO事件发生时，为了保证异步任务的及时处理，只要有任务要处理，那么就立即停止检测去处理任务。

 

三.外部线程异步执行的任务分为两种：普通任务和定时任务。这两种任务分别保存到MPSC队列和优先级队列，而优先级队列中的任务最终都会转移到MPSC队列里进行处理。

 

四.Netty每处理完64个任务才会检查一次是否超时而退出执行任务的循环。

 

(2)Reactor线程模型总结

一.NioEventLoopGroup在用户代码中被创建，默认情况下会创建两倍CPU核数个NioEventLoop。

 

二.NioEventLoop是懒启动的，bossNioEventLoop在服务端启动时启动，workerNioEventLoop在新连接接入时启动。

 

三.当CPU核数为2的幂时，为每一个新连接绑定NioEventLoop之后，都会做一个取模运算转位与运算的优化。

 

四.每个连接都对应一个Channel，每个Channel都绑定唯一一个NioEventLoop，一个NioEventLoop可能会被绑定多个Channel，每个NioEventLoop都对应一个FastThreadLocalThread线程实体和一个Selector。因此单个连接的所有操作都是在一个线程上执行的，所以是线程安全的。

 

五.每个NioEventLoop都对应一个Selector，这个Selector可以批量处理注册到它上面的Channel。

 

六.每个NioEventLoop的执行过程都包括事件检测、事件处理以及异步任务的执行。

 

七.用户线程池在对Channel进行一些操作时均为线程安全的。这是因为Netty会把外部线程的操作都封装成一个Task放入这个Channel绑定的NioEventLoop中的MPSC队列，然后在该NioEventLoop的执行过程(事件循环)的第三个过程中进行串行执行。

 

八.所以NioEventLoop的职责不仅仅是处理网络IO事件，用户自定义的普通任务和定时任务也会统一由NioEventLoop处理，从而实现线程模型的统一。

 

九.从调度层看，也不存在从NioEventLoop线程中再启动其他类型的线程用于异步执行另外的任务，从而避免了多线程并发操作和锁竞争，提升了IO线程的处理性能和调度性能。

 

(3)NioEventLoop创建启动执行的总结

一.用户在创建bossGroup和workerGroup时，NioEventLoopGroup被创建，默认不传参时会创建两倍CPU核数个NioEventLoop。

 

二.每个NioEventLoopGroup都有一个线程执行器executor和一个线程选择器chooser。线程选择器chooser用于进行线程分配，它会针对NioEventLoop的个数进行优化。

 

三.NioEventLoop在创建时会创建一个Selector和一个MPSC任务队列，创建Selector时Netty会通过反射的方式用数组去替换Selector里的两个HashSet数据结构。

 

四.Netty的NioEventLoop在首次调用execute()方法时会启动线程，这个线程是一个FastThreadLocalThread对象。启动线程后，Netty会将创建完成的线程保存到成员变量，以便能判断执行NioEventLoop里的逻辑的线程是否是这个创建好的线程。

 

五.NioEventLoop的执行逻辑在run()方法里，主要包括3部分：第一是检测IO事件，第二是处理IO事件，第三是执行异步任务。