Netty源码—3.Reactor线程模子四

大纲
5.NioEventLoop的实行总体框架
6.Reactor线程实行一次变乱轮询
7.Reactor线程处置惩罚产生IO变乱的Channel
8.Reactor线程处置惩罚使命队列之添加使命
9.Reactor线程处置惩罚使命队列之实行使命
10.NioEventLoop总结

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的这种使命实行模式来处置惩罚类似多线程数据聚合，定时上报应用。

2. //场景一：用户自定义普通任务
3. ctx.channel().eventLoop().execute(new Runnable() {
4.     @Override
5.     public void run() {
6.         ...   
7.     }
8. });
10. public final class NioEventLoop extends SingleThreadEventLoop {
11.     ...
12. }
14. public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
15.     ...
16. }
18. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
19.     //每一个NioEventLoop会有一个MPSC队列
20.     private final Queue<Runnable> taskQueue;
21.     protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
22.           boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {
23.         super(parent);
24.         this.addTaskWakesUp = addTaskWakesUp;
25.         this.maxPendingTasks = Math.max(16, maxPendingTasks);
26.         this.executor = ObjectUtil.checkNotNull(executor, "executor");
27.         //创建普通的任务队列MpscQueue
28.         taskQueue = newTaskQueue(this.maxPendingTasks);
29.         rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
30.     }
31.     ...
32.    
33.     @Override
34.     public void execute(Runnable task) {
35.         if (task == null) throw new NullPointerException("task");
36.         boolean inEventLoop = inEventLoop();
37.         //不管是外部线程还是Reactor线程执行NioEventLoop的execute()方法，都会调用NioEventLoop的addTask()方法
38.         if (inEventLoop) {
39.             addTask(task);
40.         } else {
41.             startThread();
42.             addTask(task);
43.             if (isShutdown() && removeTask(task)) {
44.                 reject();
45.             }
46.         }
47.         if (!addTaskWakesUp && wakesUpForTask(task)) wakeup(inEventLoop);
48.     }
49.    
50.     //Add a task to the task queue, or throws a RejectedExecutionException if this instance was shutdown before.
51.     protected void addTask(Runnable task) {
52.         if (task == null) throw new NullPointerException("task");
53.         if (!offerTask(task)) reject(task);
54.     }
56.     final boolean offerTask(Runnable task) {
57.         if (isShutdown()) reject();
58.         return taskQueue.offer(task);
59.     }
60.     ...
61. }

复制代码

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

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

2. //场景二：当前线程为业务线程
3. channel.write(...)
5. abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext, ResourceLeakHint {
6.     final EventExecutor executor;//一般初始化时默认为null
7.     AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name, Class<? extends ChannelHandler> handlerClass) {
8.         this.pipeline = pipeline;
9.         this.executor = executor;
10.         ...
11.     }
12.         
13.     ...
14.     private void write(Object msg, boolean flush, ChannelPromise promise) {
15.         AbstractChannelHandlerContext next = findContextOutbound();
16.         final Object m = pipeline.touch(msg, next);
17.         EventExecutor executor = next.executor();
18.         //外部线程在调用Channel的write()方法时，executor.inEventLoop()会返回false
19.         if (executor.inEventLoop()) {
20.             if (flush) {
21.                 next.invokeWriteAndFlush(m, promise);
22.             } else {
23.                 next.invokeWrite(m, promise);
24.             }
25.         } else {
26.             //将write操作封装成一个WriteTask
27.             AbstractWriteTask task;
28.             if (flush) {
29.                 task = WriteAndFlushTask.newInstance(next, m, promise);
30.             }  else {
31.                 task = WriteTask.newInstance(next, m, promise);
32.             }
33.             //调用safeExecute()来执行
34.             safeExecute(executor, task, promise, m);
35.         }
36.     }
37.    
38.     @Override
39.     public EventExecutor executor() {
40.         if (executor == null) {
41.             return channel().eventLoop();
42.         } else {
43.             return executor;
44.         }
45.     }
46.    
47.     private AbstractChannelHandlerContext findContextOutbound(int mask) {
48.         AbstractChannelHandlerContext ctx = this;
49.         EventExecutor currentExecutor = executor();
50.         do {
51.             ctx = ctx.prev;
52.         } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_OUTBOUND));
53.         return ctx;
54.     }
55.    
56.     private static void safeExecute(EventExecutor executor, Runnable runnable, ChannelPromise promise, Object msg) {
57.         try {
58.             //调用SingleThreadEventExecutor.execute()方法
59.             executor.execute(runnable);
60.         } catch (Throwable cause) {
61.             try {
62.                 promise.setFailure(cause);
63.             } finally {
64.                 if (msg != null) {
65.                     ReferenceCountUtil.release(msg);
66.                 }
67.             }
68.         }
69.     }
70.     ...
71. }
73. public class DefaultChannelPipeline implements ChannelPipeline {
74.     ...
75.     final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {
76.         private final Unsafe unsafe;
77.         HeadContext(DefaultChannelPipeline pipeline) {
78.             //传入的executor为null
79.             super(pipeline, null, HEAD_NAME, HeadContext.class);
80.             unsafe = pipeline.channel().unsafe();
81.             setAddComplete();
82.         }
83.         ...
84.     }
85.    
86.     final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
87.         TailContext(DefaultChannelPipeline pipeline) {
88.             //传入的executor为null
89.             super(pipeline, null, TAIL_NAME, TailContext.class);
90.             setAddComplete();
91.         }
92.         ...
93.     }
94.     ...
95. }
97. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
98.     //每一个NioEventLoop会有一个MPSC队列
99.     private final Queue<Runnable> taskQueue;
100.     protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
101.           boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {
102.         super(parent);
103.         this.addTaskWakesUp = addTaskWakesUp;
104.         this.maxPendingTasks = Math.max(16, maxPendingTasks);
105.         this.executor = ObjectUtil.checkNotNull(executor, "executor");
106.         //创建普通的任务队列MpscQueue
107.         taskQueue = newTaskQueue(this.maxPendingTasks);
108.         rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
109.     }
110.   
111.     ...
112.     @Override
113.     public void execute(Runnable task) {
114.         if (task == null) throw new NullPointerException("task");
115.         boolean inEventLoop = inEventLoop();
116.         //不管是外部线程还是Reactor线程执行NioEventLoop的execute()方法，都会调用NioEventLoop的addTask()方法
117.         if (inEventLoop) {
118.             addTask(task);
119.         } else {
120.             startThread();
121.             addTask(task);
122.             if (isShutdown() && removeTask(task)) {
123.                 reject();
124.             }
125.         }
126.         if (!addTaskWakesUp && wakesUpForTask(task)) wakeup(inEventLoop);
127.     }
128.    
129.     //Add a task to the task queue, or throws a RejectedExecutionException if this instance was shutdown before.
130.     protected void addTask(Runnable task) {
131.         if (task == null) throw new NullPointerException("task");
132.         if (!offerTask(task)) reject(task);
133.     }
135.     final boolean offerTask(Runnable task) {
136.         if (isShutdown()) reject();
137.         return taskQueue.offer(task);
138.     }
139.     ...
140. }

复制代码

(4)定时使命的添加
通常利用ctx.channel().eventLoop().schedule(..)自界说定时使命，此中schedule()方法会通过scheduledTaskQueue().add(task)来添加定时使命。起首scheduledTaskQueue()方法会返回一个优先级队列，然后通过该优先级队列的add()方法将定时使命对象参加到队列中。

留意，这里可以直接利用优先级队列而不消思量多线程并发题目的缘故起因如下。假如是外部线程调用schedule()方法添加定时使命，那么Netty会将添加定时使命这个逻辑封装成一个平凡的Task。这个Task的使命是一个"添加某定时使命"的使命，而不是添加某定时使命。如许，对优先级队列的访问就酿成单线程了，也就是只有Reactor线程会访问，从而不存在多线程并发题目。

2. //场景三：用户自定义定时任务，这也是用得最多的方法
3. ctx.channel().eventLoop().schedule(new Runnable() {
4.     @Override
5.     public void run() {
6.         ...   
7.     }
8. }, 60, TimeUnit.SECONDS);
10. //Abstract base class for EventExecutors that want to support scheduling.
11. public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {
12.     Queue<ScheduledFutureTask<?>> scheduledTaskQueue;
13.     ...
14.     //添加定时任务
15.     <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
16.         if (inEventLoop()) {
17.             //如果当前线程是Reactor线程，则直接往PriorityQueue中添加任务
18.             scheduledTaskQueue().add(task);
19.         } else {
20.             //如果是外部线程，则调用SingleThreadEventExecutor.execute()方法
21.             //将添加定时任务这一动作也封装成一个普通任务
22.             execute(new Runnable() {
23.                 @Override
24.                 public void run() {
25.                     scheduledTaskQueue().add(task);
26.                 }
27.             });
28.         }
29.         return task;
30.     }
31.    
32.     Queue<ScheduledFutureTask<?>> scheduledTaskQueue() {
33.         if (scheduledTaskQueue == null) {
34.             //创建定时的任务队列PriorityQueue
35.             scheduledTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>();
36.         }
37.         return scheduledTaskQueue;
38.     }
39.     ...
40. }

复制代码

(5)Netty的定时使命机制增补
一.怎样包管克制时间近来的使命优先实行
为什么定时使命要生存在优先级队列中？优先级队列的特性是会按照肯定的次序来分列内部元素，内部元素是可以比力的。由于优先级队列中的每个元素都是定时使命，以是定时使命也是可以比力的。比力的逻辑就是：先比力定时使命的克制时间，在克制时间类似的情况下再比力定时使命的添加次序也就是ID。

二.Netty的定时使命有三种实行方式
方式一：定时使命不会被重复实行
ctx.channel().eventLoop().schedule()，通报的periodNanos为0。
方式二：每隔一段时间实行一次
ctx.channel().eventLoop().scheduleAtFixedRate()，通报的periodNanos为正数。
方式三：隔类似时间再实行一次
ctx.channel().eventLoop().scheduleWithFixedDelay()，通报的periodNanos为负数。

Netty的3种定时使命的实行逻辑是通过调解下一次使命的克制时间来运行的。起首修改完下一次实行的克制时间，然后把当前使命再次参加队列，如许就能确保使命在得当的时间实行。

2. final class ScheduledFutureTask<V> extends PromiseTask<V> implements ScheduledFuture<V> {
3.     //每个定时任务都有一个唯一的ID
4.     private static final AtomicLong nextTaskId = new AtomicLong();
5.     private final long id = nextTaskId.getAndIncrement();
6.     private long deadlineNanos;
7.     //标识一个任务是否重复执行，以及以何种方式执行
8.     private final long periodNanos;
9.     ...
10.    
11.     @Override
12.     public int compareTo(Delayed o) {
13.         if (this == o) {
14.             return 0;
15.         }
17.         ScheduledFutureTask<?> that = (ScheduledFutureTask<?>) o;
18.         long d = deadlineNanos() - that.deadlineNanos();
19.         if (d < 0) {
20.             return -1;
21.         } else if (d > 0) {
22.             return 1;
23.         } else if (id < that.id) {
24.             return -1;
25.         } else if (id == that.id) {
26.             throw new Error();
27.         } else {
28.             return 1;
29.         }
30.     }
32.     public long deadlineNanos() {
33.         return deadlineNanos;
34.     }
35.    
36.     @Override
37.     public void run() {
38.         ...
39.         if (periodNanos == 0) {
40.             //1.对应schedule()方法，表示一次性任务
41.             V result = task.call();
42.             setSuccessInternal(result);
43.         } else {
44.             task.call();
45.             long p = periodNanos;
46.             if (p > 0) {
47.                 //2.对应scheduleAtFixedRate()方法，表示以固定速率执行任务
48.                 deadlineNanos += p;
49.             } else {
50.                 //3.对应scheduleWithFixedDelay()方法，表示以固定的延时执行任务
51.                 deadlineNanos = nanoTime() - p;
52.             }
53.             scheduledTaskQueue.add(this);
54.         }
55.         ...
56.     }
57.     ...
58. }

复制代码

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次盘算一下当前时间是否已超克制时间，末了判断本轮使命是否已经实行完毕。

2. //Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread.
3. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
4.     //每一个NioEventLoop会有一个MPSC队列
5.     private final Queue<Runnable> taskQueue;
6.     ...
7.    
8.     //Poll all tasks from the task queue and run them via Runnable#run() method.
9.     //This method stops running the tasks in the task queue and returns if it ran longer than timeoutNanos.
10.     protected boolean runAllTasks(long timeoutNanos) {
11.         //1.转移定时任务到MPSC队列，也就是任务聚合
12.         fetchFromScheduledTaskQueue();
13.         //从普通的任务队列(MPSC队列)里获取任务
14.         Runnable task = pollTask();
15.         if (task == null) {
16.             afterRunningAllTasks();
17.             return false;
18.         }
19.         //2.计算本轮任务执行的截止时间
20.         final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
21.         long runTasks = 0;
22.         long lastExecutionTime;
23.         //3.执行任务，通过for循环逐个执行pollTask()取出的任务
24.         for (;;) {
25.             //3.1 不抛异常地执行任务(同步阻塞)，确保任务可以安全执行
26.             safeExecute(task);
27.             //3.2 累加当前已执行的任务数
28.             runTasks ++;
29.             //3.3 每隔64次计算一下当前时间是否已经超过截止时间，因为ScheduledFutureTask.nanoTime()也挺耗时的
30.             if ((runTasks & 0x3F) == 0) {
31.                 lastExecutionTime = ScheduledFutureTask.nanoTime();
32.                 if (lastExecutionTime >= deadline) {
33.                     break;
34.                 }
35.             }
36.             //3.4 判断本轮任务是否已经执行完毕
37.             task = pollTask();
38.             if (task == null) {
39.                 lastExecutionTime = ScheduledFutureTask.nanoTime();
40.                 break;
41.             }
42.         }
43.         afterRunningAllTasks();
44.         this.lastExecutionTime = lastExecutionTime;
45.         return true;
46.     }
47.    
48.     private boolean fetchFromScheduledTaskQueue() {
49.         long nanoTime = AbstractScheduledEventExecutor.nanoTime();
50.         Runnable scheduledTask  = pollScheduledTask(nanoTime);
51.         while (scheduledTask != null) {
52.             if (!taskQueue.offer(scheduledTask)) {
53.                 scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
54.                 return false;
55.             }
56.             scheduledTask  = pollScheduledTask(nanoTime);
57.         }
58.         return true;
59.     }
60.    
61.     protected Runnable pollTask() {
62.         assert inEventLoop();
63.         return pollTaskFrom(taskQueue);
64.     }
66.     protected final Runnable pollTaskFrom(Queue<Runnable> taskQueue) {
67.         for (;;) {
68.             Runnable task = taskQueue.poll();
69.             if (task == WAKEUP_TASK) {
70.                 continue;
71.             }
72.             return task;
73.         }
74.     }
75.     ...
76. }
78. //Abstract base class for EventExecutors that want to support scheduling.
79. public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {
80.     Queue<ScheduledFutureTask<?>> scheduledTaskQueue;
81.     ...
82.     //Return the Runnable which is ready to be executed with the given nanoTime.
83.     //You should use #nanoTime() to retrieve the the correct nanoTime.
84.     protected final Runnable pollScheduledTask(long nanoTime) {
85.         assert inEventLoop();
86.         Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
87.         ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
88.         if (scheduledTask == null) {
89.             return null;
90.         }
91.         if (scheduledTask.deadlineNanos() <= nanoTime) {
92.             scheduledTaskQueue.remove();
93.             return scheduledTask;
94.         }
95.         return null;
96.     }
97.     ...
98. }
100. //Abstract base class for EventExecutor implementations.
101. public abstract class AbstractEventExecutor extends AbstractExecutorService implements EventExecutor {
102.     ...
103.     //Try to execute the given Runnable and just log if it throws a Throwable.
104.     protected static void safeExecute(Runnable task) {
105.         try {
106.             task.run();//同步执行任务
107.         } catch (Throwable t) {
108.             logger.warn("A task raised an exception. Task: {}", task, t);
109.         }
110.     }
111.     ...
112. }

复制代码

(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变乱，第三是实行异步使命。