netty5笔记-线程模型3-EventLoop

落日映苍穹つ 2022-05-28 01:38 235阅读 0赞

NioEventLoop相对NioEventLoopGroup来说就复杂很多了，需要一定的耐心来看这篇文章。

首先从NioEventLoop的启动讲起，对于线程池来说，启动一般都是从第一个任务的添加开始的。经过跟踪，找到execute()方法在SingleThreadEventExecutor类中：

**\[java\]**  [view plain][]  [copy][view plain]

1.      public void execute(Runnable task) \{  
2.          if (task == null) \{  
3.              throw new NullPointerException("task");  
4.          \}  
5.    
6.          // inEventLoop表示启动线程与当前线程相同，相同表示已经启动，不同则有两种可能：未启动或者线程不同  
7.          boolean inEventLoop = inEventLoop();  
8.          if (inEventLoop) \{  
9.              // 运行中则直接添加任务到队列中  
10.             addTask(task);  
11.         \} else \{  
12.             // 尝试启动任务  
13.             startExecution();  
14.             // 将任务加到任务队列taskQueue中  
15.             addTask(task);  
16.             // 发现已经关闭则移除任务并拒绝  
17.             if (isShutdown() && removeTask(task)) \{  
18.                 reject();  
19.             \}  
20.         \}  
21.   
22.         if (!addTaskWakesUp && wakesUpForTask(task)) \{  
23.             // 唤醒执行线程  
24.             wakeup(inEventLoop);  
25.         \}  
26.     \}  
27.     private void startExecution() \{  
28.         // 未启动的状态下才进行启动  
29.         if (STATE\_UPDATER.get(this) == ST\_NOT\_STARTED) \{  
30.             if (STATE\_UPDATER.compareAndSet(this, ST\_NOT\_STARTED, ST\_STARTED)) \{  
31.                 // 增加一个定时任务，该任务将定时任务队列中的已取消任务从队列中移除，该任务每间隔1秒执行1次  
32.                 schedule(new ScheduledFutureTask<Void>(  
33.                         this, Executors.<Void>callable(new PurgeTask(), null),  
34.                         ScheduledFutureTask.deadlineNanos(SCHEDULE\_PURGE\_INTERVAL), -SCHEDULE\_PURGE\_INTERVAL));  
35.                 // 开始执行  
36.                 scheduleExecution();  
37.             \}  
38.         \}  
39.     \}  
40.       
41.     // 如果已经关闭了，则不能再加任务，否则加入到任务队列中  
42.     protected void addTask(Runnable task) \{  
43.         if (task == null) \{  
44.             throw new NullPointerException("task");  
45.         \}  
46.         if (isShutdown()) \{  
47.             reject();  
48.         \}  
49.         taskQueue.add(task);  
50.     \}

简单的部分就不用讲了，我们来看看两个可能会让人疑惑的点：

1、scheduleExecution()

这个方法是将asRunnable提交到executor，由于线程池的线程数与EventExecutor的个数相同，所以可以保证每次asRunnable都能及时处理， asRunnable逻辑比较简单，执行所在类中的run方法，这个run方法是个抽象方法，它的实现有几个要求要满足：

a、run方法中只执行一定量的任务。如果执行太多，或者一直执行不跳出，那么后期netty中期望引入的fork/jion框架stealing机制就会失效或者大打折扣；

b、run方法执行完一定量任务后，本次任务完成，此时需要调用scheduleExecution()，否则该EventExecutor后面的任务将无法进行；

c、基于b中子类必须调用scheduleExecution()的要求，任务的执行必须使用try catch方式。如果不这样的话，发生任何异常都会导致EventExecutor关闭，里面的所有任务都将被清理。

另一个需要注意的点是scheduleExecution方法在执行asRunnable前将thread置为null了，该thread表示EventLoop所在线程，由于executor.execute的执行并不能保证是哪个Thread来执行，因此先把thread置为null，等进行asRunnable的run方法后再次设置thread为Thread.currentThread。

**\[java\]**  [view plain][]  [copy][view plain]

1.      protected final void scheduleExecution() \{  
2.          updateThread(null);  
3.          executor.execute(asRunnable);  
4.      \}  
5.    
6.      private final Runnable asRunnable = new Runnable() \{  
7.          @Override  
8.          public void run() \{  
9.              updateThread(Thread.currentThread());  
10.             // lastExecutionTime must be set on the first run  
11.             // in order for shutdown to work correctly for the  
12.             // rare case that the eventloop did not execute  
13.             // a single task during its lifetime.  
14.             if (firstRun) \{  
15.                 firstRun = false;  
16.                 updateLastExecutionTime();  
17.             \}  
18.             try \{  
19.                 SingleThreadEventExecutor.this.run();  
20.             \} catch (Throwable t) \{  
21.                 // 发生异常则关闭整个EventExecutor  
22.                 logger.warn("Unexpected exception from an event executor: ", t);  
23.                 cleanupAndTerminate(false);  
24.             \}  
25.      \}

下面我们写一段简单的代码来看看，一个简单的异常是如何使整个EventExecutor挂掉的:

**\[java\]**  [view plain][]  [copy][view plain]

1.      public static void main(String\[\] args) throws Exception \{  
2.          DefaultEventExecutorGroup group = new DefaultEventExecutorGroup(1);  
3.          group.execute(new Runnable() \{  
4.              @Override  
5.              public void run() \{  
6.                  System.out.println("a point");  
7.                  throw new RuntimeException("runtime error");  
8.              \}  
9.          \});  
10.         Thread.sleep(100);  
11.         group.execute(new Runnable() \{  
12.             @Override  
13.             public void run() \{  
14.                 System.out.println("b point");  
15.                 throw new RuntimeException("runtime error");  
16.             \}  
17.         \});  
18.           
19.         Thread.sleep(10000);  
20.         group.close();  
21.     \}  
22.   
23.     // DefaultEventExecutorGroup 的问题在于其run方法并没有捕获异常，不知道现在修复没有。注意，4.x由于线程模型不同，类似的代码并不会有太大什么问题。  
24.     protected void run() \{  
25.         Runnable task = takeTask();  
26.         if (task != null) \{  
27.             // 正确的做法是捕获task.run的异常  
28.             task.run();  
29.             updateLastExecutionTime();  
30.         \}  
31.   
32.         if (confirmShutdown()) \{  
33.             cleanupAndTerminate(true);  
34.         \} else \{  
35.             scheduleExecution();  
36.         \}  
37.     \}

2、wakeup

先回过头看看第一段代码中调用wakeup的地方，其中addTaskWakesUp的表示添加任务时是否会唤醒线程。啥意思呢，比如一个线程里执行了一个阻塞方法 BlockingQueue.take()，该方法在没有获取到数据的时候一直阻塞，要想恢复，往该BlockingQueue中加一个对象就可以了，线程恢复了执行，就可以进行其他判断了（如线程是否被关闭之类的判断）。DefaultEventExecutor传入的addTaskWakesUp=true, 因为它能阻塞的地方就在BlockingQueue.take()，因此加入一个任务可以唤醒线程，这里就为true。 而NioEventLoop阻塞的地方在selector.select，加入task也无法立即唤醒线程，因此addTaskWakesUp=false。

**\[java\]**  [view plain][]  [copy][view plain]

1.  if (!addTaskWakesUp && wakesUpForTask(task)) \{  
2.      wakeup(inEventLoop);  
3.  \}

我们来看看针对DefaultEventExecutor和NioEventLoop，wakeup方法有何不同：

**\[java\]**  [view plain][]  [copy][view plain]

1.      // NioEventLoop通过selector.wakeup()来使阻塞在selector.select上的方法恢复  
2.      protected void wakeup(boolean inEventLoop) \{  
3.          if (!inEventLoop && wakenUp.compareAndSet(false, true)) \{  
4.              selector.wakeup();  
5.          \}  
6.      \}  
7.      // DefaultEventExecutor则通过往queue里面加一个空的Runnable来是阻塞的take方法恢复，这个也是默认的实现。  
8.      protected void wakeup(boolean inEventLoop) \{  
9.          if (!inEventLoop || STATE\_UPDATER.get(this) == ST\_SHUTTING\_DOWN) \{  
10.             taskQueue.add(WAKEUP\_TASK);  
11.         \}  
12.     \}  
13.       
14.     private static final Runnable WAKEUP\_TASK = new Runnable() \{  
15.         @Override  
16.         public void run() \{  
17.             // Do nothing.  
18.         \}  
19.     \};

通常在整个EventExecutor关闭或者添加一个任务时都需要调用唤醒方法。 如果你自己实现的子类里还有其他方法会阻塞，你就需要想办法来恢复线程。

ok，下面开始讲run方法，也是大家最容易感兴趣的方法。SingleThreadEventExecutor的run方法为抽象方法，具体的实现在子类中，那么我们回到NioEventLoop：

**\[java\]**  [view plain][]  [copy][view plain]

1.      protected void run() \{  
2.          // 每次进来都将wokenUp设为false，这样如果有新任务提交，会触发一次selector.wakeup，这样即使进入下面的select(oldWakenUp)分支  
3.          // 也能保证新提交任务能及时执行  
4.          boolean oldWakenUp = wakenUp.getAndSet(false);  
5.          try \{  
6.              if (hasTasks()) \{  
7.                  // 当有任务时为了保证任务及时执行采用不阻塞的selectNow获取准备好I/O的连接  
8.                  selectNow();  
9.              \} else \{  
10.                 // 当无任务时采用阻塞等待的方式获取连接  
11.                 select(oldWakenUp);  
12.                 if (wakenUp.get()) \{  
13.                     selector.wakeup();  
14.                 \}  
15.             \}  
16.   
17.             cancelledKeys = 0;  
18.             needsToSelectAgain = false;  
19.             final int ioRatio = this.ioRatio;  
20.             if (ioRatio == 100) \{  
21.                 processSelectedKeys();  
22.                 runAllTasks();  
23.             \} else \{  
24.                 final long ioStartTime = System.nanoTime();  
25.   
26.                 processSelectedKeys();  
27.   
28.                 final long ioTime = System.nanoTime() - ioStartTime;  
29.                 // 根据处理selectKeys的时间 和 ioRatio计算得到处理普通task的时间  
30.                 runAllTasks(ioTime \* (100 - ioRatio) / ioRatio);  
31.             \}  
32.   
33.             // 如果被关闭了，则关闭所有连接(closeAll)，并且完成对应的清理任务  
34.             if (isShuttingDown()) \{  
35.                 closeAll();  
36.                 if (confirmShutdown()) \{  
37.                     cleanupAndTerminate(true);  
38.                     return;  
39.                 \}  
40.             \}  
41.         \} catch (Throwable t) \{  
42.             ...  
43.         \}  
44.   
45.         scheduleExecution();  
46.     \}

select(oldWakeUp)与selectNow都是获取已经准备好的连接，不同的是select(oldWakeUp)会产生阻塞，其处理如下：

1、获取定时任务中最近执行的任务，并根据这个时间确定select(timeout)的timeout值，如下个定时任务1秒后执行，则select(1000), 即等待1秒后不管有没有准备好的连接都会返回。 由于EventLoop启动时加入了一个每秒执行一次的任务，这里select最多不会超过1秒， 需要注意的是由于加入定时任务是并不会调用selector.wakeup()，因此执行线程进入select(timeout)后，如果其他线程加入了定时任务且时间小于timeout，就无法及时执行，不过误差小于1秒问题不大。  顺便提醒下nio的异步阻塞的“阻塞”就是指select(timeout)这里；

2、如果发现有定时任务已经可以执行了，则直接selectNow()后返回；

3、java早期的nio bug会导致cpu 100%, 此时select(timeout)不会阻塞直接返回0， 在netty中判断方式为在很短时间内（小于1秒）完成了多次（默认512）select(timeout)，则发生了该bug，此时进行rebuildSelector来消除bug。精简后代码如下：

**\[java\]**  [view plain][]  [copy][view plain]

1.            ......  
2.            for (;;) \{  
3.                  long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;  
4.                  int selectedKeys = selector.select(timeoutMillis);  
5.                  selectCnt ++;  
6.                  long time = System.nanoTime();  
7.                  if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) \{  
8.                      // 正常阻塞则将selectCnt置为1，否则selectCnt会一直累加知道进入下面一个分支  
9.                      selectCnt = 1;  
10.                 \} else if (selectCnt >= 512) \{  
11.                     rebuildSelector();  
12.                     ......  
13.                 \}......  
14.             \}

接着来看看processSelectedKeys。从方法名可以看出这里主要是处理前面select获取到的已经准备ok的连接。 根据优化情况选择processSelectedKeysPlain和processSelectedKeysOptimized方法，两个方法代码类似。

**\[java\]**  [view plain][]  [copy][view plain]

1.  private void processSelectedKeysOptimized(SelectionKey\[\] selectedKeys) \{  
2.          for (int i = 0;; i ++) \{  
3.              // 在获取所有key(即flip）时会将未最后一个有效key的下一个位置值为null，因此碰到null，说明所有有效的key已经获取完  
4.              final SelectionKey k = selectedKeys\[i\];  
5.              if (k == null) \{  
6.                  break;  
7.              \}  
8.              // null out entry in the array to allow to have it GC'ed once the Channel close  
9.              // See https://github.com/netty/netty/issues/2363  
10.             selectedKeys\[i\] = null;  
11.   
12.             final Object a = k.attachment();  
13.   
14.             // key关联两种不同类型的对象，一种是AbstractNioChannel，一种是NioTask  
15.             if (a instanceof AbstractNioChannel) \{  
16.                 processSelectedKey(k, (AbstractNioChannel) a);  
17.             \} else \{  
18.                 NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;  
19.                 processSelectedKey(k, task);  
20.             \}  
21.   
22.             // 如果需要重新select则重置当前数据  
23.             if (needsToSelectAgain) \{  
24.                 // null out entries in the array to allow to have it GC'ed once the Channel close  
25.                 // See https://github.com/netty/netty/issues/2363  
26.                 for (;;) \{  
27.                     if (selectedKeys\[i\] == null) \{  
28.                         break;  
29.                     \}  
30.                     selectedKeys\[i\] = null;  
31.                     i++;  
32.                 \}  
33.   
34.                 selectAgain();  
35.                 // Need to flip the optimized selectedKeys to get the right reference to the array  
36.                 // and reset the index to -1 which will then set to 0 on the for loop  
37.                 // to start over again.  
38.                 //  
39.                 // See https://github.com/netty/netty/issues/1523  
40.                 selectedKeys = this.selectedKeys.flip();  
41.                 i = -1;  
42.             \}  
43.         \}  
44.     \}

上面的处理过程中有一个needsToSelectAgain，什么情况下会触发这个条件呢。当多个channel从selector中撤销注册时，由于很多数据无效了（默认为256），需要重新处理：

**\[java\]**  [view plain][]  [copy][view plain]

1.  void cancel(SelectionKey key) \{  
2.      key.cancel();  
3.      cancelledKeys ++;  
4.      if (cancelledKeys >= CLEANUP\_INTERVAL) \{  
5.          cancelledKeys = 0;  
6.          needsToSelectAgain = true;  
7.      \}  
8.  \}

一个selectedKey的attachment可能对应AbstractNioChannel和NioTask两种对象。第一种很好理解，就是我们常用的netty nio连接。另一个NioTask则是作者给我们留的一个接口，他可以允许开发者自己去实现一个非netty AbstractNioChannel的SelectableChannel， 对于这种对象，准备好数据后会调用对象的NioTask.channelReady方法，由开发者自己实现对应的方法。如果你想要一个NioTask的例子，很遗憾的告诉你我没有，也不想写，连netty的开发者都说“你要是实现了请告诉我”，当然，他说的是英文！ 再看看Channel的处理：

**\[java\]**  [view plain][]  [copy][view plain]

1.  private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) \{  
2.      final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();  
3.      if (!k.isValid()) \{  
4.          // close the channel if the key is not valid anymore  
5.          unsafe.close(unsafe.voidPromise());  
6.          return;  
7.      \}  
8.    
9.      try \{  
10.         int readyOps = k.readyOps();  
11.         // 如果准备好READ或ACCEPT则触发channel.unsafe().read()  
12.         if ((readyOps & (SelectionKey.OP\_READ | SelectionKey.OP\_ACCEPT)) != 0 || readyOps == 0) \{  
13.             // 这里的操作与channel相关，不是本文重点，但以后会介绍  
14.             unsafe.read();  
15.             if (!ch.isOpen()) \{  
16.                 // 如果已经关闭，则不需要再处理该channel的其他事件，直接返回  
17.                 return;  
18.             \}  
19.         \}  
20.         if ((readyOps & SelectionKey.OP\_WRITE) != 0) \{  
21.             // 如果准备好了WRITE则将缓冲区中的数据发送出去，如果缓冲区中数据都发送完成，则清除之前关注的OP\_WRITE标记  
22.             ch.unsafe().forceFlush();  
23.         \}  
24.         if ((readyOps & SelectionKey.OP\_CONNECT) != 0) \{  
25.             // 需要移除OP\_CONNECT否则Selector.select(timeout)可能会出现cpu 100%  
26.             // See https://github.com/netty/netty/issues/924  
27.             int ops = k.interestOps();  
28.             ops &= ~SelectionKey.OP\_CONNECT;  
29.             k.interestOps(ops);  
30.   
31.             unsafe.finishConnect();  
32.         \}  
33.     \} catch (CancelledKeyException ignored) \{  
34.         unsafe.close(unsafe.voidPromise());  
35.     \}  
36. \}

runAllTasks主要分成两步：

1、从定时任务队列拉取到执行时间的任务到任务队列；

2、循环从任务队列里取数据，知道队列为空。正常情况下ioRatio是非100的，所以for这部分的执行是有时间限制的，具体代码见runAllTasks(long timeoutNanos)，这里就不再贴出了。

还记得之前介绍的wakeup方法吗，NioEventLoop只对selector进行了wakeup，而没有对队列进行wakeup，因为下面的pollTask是采用的非阻塞方式。

**\[java\]**  [view plain][]  [copy][view plain]

1.  protected boolean runAllTasks() \{  
2.      fetchFromScheduledTaskQueue();  
3.      Runnable task = pollTask();  
4.      if (task == null) \{  
5.          return false;  
6.      \}  
7.    
8.      for (;;) \{  
9.          try \{  
10.             task.run();  
11.         \} catch (Throwable t) \{  
12.             logger.warn("A task raised an exception.", t);  
13.         \}  
14.           
15.         task = pollTask();  
16.         if (task == null) \{  
17.             lastExecutionTime = ScheduledFutureTask.nanoTime();  
18.             return true;  
19.         \}  
20.     \}  
21. \}

到这里run方法介绍得差不多了，剩下一段对关闭状态的处理。 关闭的处理我们可以先看shutdownGracefully，netty号称实现优雅关闭，那么它是如何优雅的？

1、如果ST\_NOT\_STARTED或者ST\_STARTED尝试将状态切换为ST\_SHUTTING\_DOWN，如果被别的线程抢先执行了，则此线程直接返回Future等待结果即可；

2、切换状态成功的线程可以进行后面的逻辑： 如果线程未启动则发起一次scheduleExecution，让EventLoop进行后面的清理逻辑。

3、如果线程在执行中则进行wakeup唤起阻塞的线程。

3、EventLoop执行run方法的倒数第二部分：判断状态被置为关闭，进行最后的清理工作。

**\[java\]**  [view plain][]  [copy][view plain]

1.  public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) \{  
2.            
3.          boolean inEventLoop = inEventLoop();  
4.          boolean wakeup;  
5.          int oldState;  
6.          // 尝试切换EventLoop状态，如果竞争失败则返回Future等待结果  
7.          for (;;) \{  
8.              if (isShuttingDown()) \{  
9.                  return terminationFuture();  
10.             \}  
11.             int newState;  
12.             wakeup = true;  
13.             oldState = STATE\_UPDATER.get(this);  
14.             if (inEventLoop) \{  
15.                 newState = ST\_SHUTTING\_DOWN;  
16.             \} else \{  
17.                 switch (oldState) \{  
18.                     case ST\_NOT\_STARTED:  
19.                     case ST\_STARTED:  
20.                         newState = ST\_SHUTTING\_DOWN;  
21.                         break;  
22.                     default:  
23.                         newState = oldState;  
24.                         wakeup = false;  
25.                 \}  
26.             \}  
27.             // 由于newState和oldState可能相同，这里可能执行多次，但是没有关系，在关闭状态下即使这里成功了，也不满足执行scheduleExecution和wakeup的条件  
28.             if (STATE\_UPDATER.compareAndSet(this, oldState, newState)) \{  
29.                 break;  
30.             \}  
31.         \}  
32.         gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);  
33.         gracefulShutdownTimeout = unit.toNanos(timeout);  
34.           
35.         if (oldState == ST\_NOT\_STARTED) \{  
36.             scheduleExecution();  
37.         \}  
38.         // 如果之前的状态是运行中，需要进行一次唤醒，防止一直阻塞或者阻塞时间很长  
39.         if (wakeup) \{  
40.             wakeup(inEventLoop);  
41.         \}  
42.   
43.         return terminationFuture();  
44.     \}

清理包括以下几步：

1、关闭当前EventLoop下的selector维护的所有连接，包括AbstractNioChannel和NioTask, 对应closeAll();

2、取消所有定时任务并清空定时任务队列，所有未执行的非定时任务执行完毕, 所有shutdownHook执行完毕，对应confirmShutdown();

3、如果confirmShutdown()失败（返回false）则进入下一轮run继续尝试，否则进行cleanupAndTerminate方法，循环调用confirmShutdown()直到所有任务执行完，将状态设为ST\_TERMINATED， 将selector关闭。 confirmShutdown返回失败的场景：confirmShutdown方法中成功执行了一个任务则返回失败，而由于shutdownGracefully只是将状态设为ST\_SHUTTING\_DOWN，还可以往队列中添加任务，因此这里有失败的可能。

最后我们做个简单的总结：

1、通过execute方法触发运行，运行方式为使用executor.execute执行asRunnable， asRunnable执行EventExecutor类中的run方法；

2、子类通过实现run方法来定制自己的功能。在NioEventLoop中，执行一批操作的过程如下：

2.1 从selector中取出准备好的连接，处理这批连接的读、写事件或者NioTask中的channelReady方法

2.2 处理非I/O事件

2.3 判断状态是否为ST\_SHUTTING\_DOWN，如果是则进行资源清理操作，包括关闭连接、取消定时任务、处理剩余的非定时任务、处理shutdownHook, 关闭selector

2.4 如果状态不为ST\_SHUTTING\_DOWN，再次调用executor.execute方法执行asRunnable，如此循环。 同一个EventExecutor中执行完一批操作才会触发下一批操作，因此依然是线程安全的；

3、通过shutdownGracefully关闭，主要是设置关闭状态，并触发run方法的执行（执行到2.3），通过这种方式让对应的EventExecutor生命周期自然终止。

[view plain]: https://blog.csdn.net/youaremoon/article/details/50311341#