Netty源码阅读之NioEventLoop简析

梦里梦外; 2023-02-20 08:19 141阅读 0赞 
  1.  在Netty中NioEventLoop以及NioEventLoopGroup是很重要的两个类，而NioEventLoopGroup主要是对NioEventLoop进行管理；首先来看一下这两个类的关系图（错综复杂）：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbGlzaGFkYWE_size_16_color_FFFFFF_t_70

  1.                                                                                              图 1

1. NioEventLoopGroup**初始化流程**

通过分析NioEventLoopGroup的构造方法的调用栈我们能够看到在io.netty.channel.MultithreadEventLoopGroup的构造方法中进行了创建：

当未指定具体的线程数目的时候，Netty会提出一个默认的线程数：DEFAULT_EVENT_LOOP_THREADS

  1. protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
  2.         super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
  3.     }

而该数值在同一类下的静态代码块中进行了设置：

  1. static {
  2.         DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
  3.                 "io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
  5.         if (logger.isDebugEnabled()) {
  6.             logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
  7.         }
  8.     }

显而易见，默认的线程数量为2*cpu数目。

继续深入，打开io.netty.util.concurrent.MultithreadEventExecutorGroup这个类，查看其构造方法：

  1. protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
  2.                                             EventExecutorChooserFactory chooserFactory, Object... args) {
  3.         if (nThreads <= 0) {
  4.             throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
  5.         }
  7.         if (executor == null) {
  8.             executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());//创建线程执行器
  9.         }
  11.         //构造NioEventLoop的过程
  12.         children = new EventExecutor[nThreads];
  14.         for (int i = 0; i < nThreads; i ++) {
  15.             boolean success = false;
  16.             try {
  17.                 children[i] = newChild(executor, args);
  18.                 success = true;
  19.             } catch (Exception e) {
  20.                 // TODO: Think about if this is a good exception type
  21.                 throw new IllegalStateException("failed to create a child event loop", e);
  22.             } finally {
  23.                 if (!success) {
  24.                     for (int j = 0; j < i; j ++) {
  25.                         children[j].shutdownGracefully();
  26.                     }
  28.                     for (int j = 0; j < i; j ++) {
  29.                         EventExecutor e = children[j];
  30.                         try {
  31.                             while (!e.isTerminated()) {
  32.                                 e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
  33.                             }
  34.                         } catch (InterruptedException interrupted) {
  35.                             // Let the caller handle the interruption.
  36.                             Thread.currentThread().interrupt();
  37.                             break;
  38.                         }
  39.                     }
  40.                 }
  41.             }
  42.         }
  44.         chooser = chooserFactory.newChooser(children);//生成线程选择器
  46.         final FutureListener<Object> terminationListener = new FutureListener<Object>() {
  47.             @Override
  48.             public void operationComplete(Future<Object> future) throws Exception {
  49.                 if (terminatedChildren.incrementAndGet() == children.length) {
  50.                     terminationFuture.setSuccess(null);
  51.                 }
  52.             }
  53.         };
  55.         for (EventExecutor e: children) {
  56.             e.terminationFuture().addListener(terminationListener);
  57.         }
  59.         Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
  60.         Collections.addAll(childrenSet, children);
  61.         readonlyChildren = Collections.unmodifiableSet(childrenSet);
  62.     }

通过上面的代码流程可知，首先是创建线程执行器，

线程执行器中传入一个默认的线程工厂：newDefaultThreadFactory，在线程工厂中进行nio线程的创建并进行线程的命名：

  1. public static String toPoolName(Class<?> poolType) {
  2.         if (poolType == null) {
  3.             throw new NullPointerException("poolType");
  4.         }
  6.         String poolName = StringUtil.simpleClassName(poolType);
  7.         switch (poolName.length()) {
  8.             case 0:
  9.                 return "unknown";
  10.             case 1:
  11.                 return poolName.toLowerCase(Locale.US);
  12.             default:
  13.                 if (Character.isUpperCase(poolName.charAt(0)) && Character.isLowerCase(poolName.charAt(1))) {
  14.                     return Character.toLowerCase(poolName.charAt(0)) + poolName.substring(1);
  15.                 } else {
  16.                     return poolName;
  17.                 }
  18.         }
  19.     }

线程名称类似于：nioEventLoop-x-x这种形式；同时，将根据线程数目创建一个同等容量的EventExecutor数组，数组中通过newChild()方法塞入一个EventExecutor对象，当然这只是一个抽象方法，具体的实现根据不同的类来决定，若在这个过程中有一个线程发生了异常，则会从当前的这个线程开始，将前面从第一个线程开始，关闭对应的线程执行器；之后再初始化线程选择器工厂，并通过轮询算法来处理本次的所有EventLoop事件,加入线程工厂的时候，采用了策略模式，会有一个2次幂的判断，如果上述的数组长度为2的幂次方，那么选用PowerOfTowEventExecutorChooser(executors)

，否则将选用GenericEventExecutorChooser(executors)

  1. public EventExecutorChooser newChooser(EventExecutor[] executors) {
  2.         if (isPowerOfTwo(executors.length)) {
  3.             return new PowerOfTowEventExecutorChooser(executors);
  4.         } else {
  5.             return new GenericEventExecutorChooser(executors);
  6.         }
  7.     }

而这两种方式主要在遍历数组的时候存在区别，当为2的次幂的时候，采用如下方式进行遍历：

  1. public EventExecutor next() {
  2.             return executors[idx.getAndIncrement() & executors.length - 1];
  3.         }

反之则采用如下的方式进行数组的遍历：

  1. public EventExecutor next() {
  2.             return executors[Math.abs(idx.getAndIncrement() % executors.length)];
  3.         }

由上面的分析，我们可以得出Netty中的EventLoop处理关系图：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbGlzaGFkYWE_size_16_color_FFFFFF_t_70 1

  1.                                                                                         图 2

2. NioEventLoop启动逻辑

启动的入口为：io.netty.bootstrap.AbstractBootstrap#doBind0()方法：

  1. private static void doBind0(
  2.             final ChannelFuture regFuture, final Channel channel,
  3.             final SocketAddress localAddress, final ChannelPromise promise) {
  5.         // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
  6.         // the pipeline in its channelRegistered() implementation.
  7.         channel.eventLoop().execute(new Runnable() {
  8.             @Override
  9.             public void run() {
  10.                 if (regFuture.isSuccess()) {
  11.                     channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
  12.                 } else {
  13.                     promise.setFailure(regFuture.cause());
  14.                 }
  15.             }
  16.         });
  17.     }

主要是进行端口的绑定。

接着往下查看execute()方法：

  1. public void execute(Runnable task) {
  2.         if (task == null) {
  3.             throw new NullPointerException("task");
  4.         }
  6.         boolean inEventLoop = inEventLoop();
  7.         if (inEventLoop) {
  8.             addTask(task);
  9.         } else {
  10.             startThread();
  11.             addTask(task);
  12.             if (isShutdown() && removeTask(task)) {
  13.                 reject();
  14.             }
  15.         }
  17.         if (!addTaskWakesUp && wakesUpForTask(task)) {
  18.             wakeup(inEventLoop);
  19.         }
  20.     }

在execute()方法中，首先通过inEventLoop()方法判断当前的线程是否是在eventLoop中，值得注意的是，每一个NioEventLoop都维护着一个taskQueue，读写任务都将被丢进这个队列中进行维护：

  1. @Override
  2.     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
  3.         // This event loop never calls takeTask()
  4.         return PlatformDependent.newMpscQueue(maxPendingTasks);
  5.     }

这是Netty实现异步串行无锁化的关键；回归正题，如果已经在evetLoop中了，那么直接将当前的任务添加到任务队列中，否则将执行doStartThread()方法：

  1. private void doStartThread() {
  2.         assert thread == null;
  3.         executor.execute(new Runnable() {
  4.             @Override
  5.             public void run() {
  6.                 thread = Thread.currentThread();
  7.                 if (interrupted) {
  8.                     thread.interrupt();
  9.                 }
  11.                 boolean success = false;
  12.                 updateLastExecutionTime();
  13.                 try {
  14.                     SingleThreadEventExecutor.this.run();
  15.                     success = true;
  16.                 } catch (Throwable t) {
  17.                     logger.warn("Unexpected exception from an event executor: ", t);
  18.                 } finally {
  19.                     for (;;) {
  20.                         int oldState = STATE_UPDATER.get(SingleThreadEventExecutor.this);
  21.                         if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
  22.                                 SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
  23.                             break;
  24.                         }
  25.                     }
  27.                     // Check if confirmShutdown() was called at the end of the loop.
  28.                     if (success && gracefulShutdownStartTime == 0) {
  29.                         logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +
  30.                                 SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called " +
  31.                                 "before run() implementation terminates.");
  32.                     }
  34.                     try {
  35.                         // Run all remaining tasks and shutdown hooks.
  36.                         for (;;) {
  37.                             if (confirmShutdown()) {
  38.                                 break;
  39.                             }
  40.                         }
  41.                     } finally {
  42.                         try {
  43.                             cleanup();
  44.                         } finally {
  45.                             STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
  46.                             threadLock.release();
  47.                             if (!taskQueue.isEmpty()) {
  48.                                 logger.warn(
  49.                                         "An event executor terminated with " +
  50.                                                 "non-empty task queue (" + taskQueue.size() + ')');
  51.                             }
  53.                             terminationFuture.setSuccess(null);
  54.                         }
  55.                     }
  56.                 }
  57.             }
  58.         });
  59.     }

通过SingleThreadEventExecutor.this.run()方法，Netty中的channel将不断轮询处理channel事件：

  1. protected void run() {
  2.         for (;;) {
  3.             try {
  4.                 switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
  5.                     case SelectStrategy.CONTINUE:
  6.                         continue;
  7.                     case SelectStrategy.SELECT:
  8.                         select(wakenUp.getAndSet(false));
  9.                         if (wakenUp.get()) {
  10.                             selector.wakeup();
  11.                         }
  12.                     default:
  13.                         // fallthrough
  14.                 }
  16.                 cancelledKeys = 0;
  17.                 needsToSelectAgain = false;
  18.                 final int ioRatio = this.ioRatio;
  19.                 if (ioRatio == 100) {
  20.                     try {
  21.                         processSelectedKeys();
  22.                     } finally {
  23.                         // Ensure we always run tasks.
  24.                         runAllTasks();
  25.                     }
  26.                 } else {
  27.                     final long ioStartTime = System.nanoTime();
  28.                     try {
  29.                         processSelectedKeys();
  30.                     } finally {
  31.                         // Ensure we always run tasks.
  32.                         final long ioTime = System.nanoTime() - ioStartTime;
  33.                         runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
  34.                     }
  35.                 }
  36.             } catch (Throwable t) {
  37.                 handleLoopException(t);
  38.             }
  39.             // Always handle shutdown even if the loop processing threw an exception.
  40.             try {
  41.                 if (isShuttingDown()) {
  42.                     closeAll();
  43.                     if (confirmShutdown()) {
  44.                         return;
  45.                     }
  46.                 }
  47.             } catch (Throwable t) {
  48.                 handleLoopException(t);
  49.             }
  50.         }
  51.     }

在事件循环中不仅需要处理IO事件也需要处理非IO事件，IO事件处理通过processSelectedKeys方法来进行，而非IO事件通过runAllTasks()方法进行处理，IO事件以及非IO事件的默认占比各为50%，值得注意的是：SelectStrategy.SELECT这种情况：

  1. private void select(boolean oldWakenUp) throws IOException {
  2.         Selector selector = this.selector;
  3.         try {
  4.             int selectCnt = 0;
  5.             long currentTimeNanos = System.nanoTime();
  6.             long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
  7.             for (;;) {
  8.                 long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
  9.                 if (timeoutMillis <= 0) {
  10.                     if (selectCnt == 0) {
  11.                         selector.selectNow();
  12.                         selectCnt = 1;
  13.                     }
  14.                     break;
  15.                 }
  17.                 // If a task was submitted when wakenUp value was true, the task didn't get a chance to call
  18.                 // Selector#wakeup. So we need to check task queue again before executing select operation.
  19.                 // If we don't, the task might be pended until select operation was timed out.
  20.                 // It might be pended until idle timeout if IdleStateHandler existed in pipeline.
  21.                 if (hasTasks() && wakenUp.compareAndSet(false, true)) {
  22.                     selector.selectNow();
  23.                     selectCnt = 1;
  24.                     break;
  25.                 }
  27.                 int selectedKeys = selector.select(timeoutMillis);
  28.                 selectCnt ++;
  30.                 if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
  31.                     // - Selected something,
  32.                     // - waken up by user, or
  33.                     // - the task queue has a pending task.
  34.                     // - a scheduled task is ready for processing
  35.                     break;
  36.                 }
  37.                 if (Thread.interrupted()) {
  38.                     // Thread was interrupted so reset selected keys and break so we not run into a busy loop.
  39.                     // As this is most likely a bug in the handler of the user or it's client library we will
  40.                     // also log it.
  41.                     //
  42.                     // See https://github.com/netty/netty/issues/2426
  43.                     if (logger.isDebugEnabled()) {
  44.                         logger.debug("Selector.select() returned prematurely because " +
  45.                                 "Thread.currentThread().interrupt() was called. Use " +
  46.                                 "NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
  47.                     }
  48.                     selectCnt = 1;
  49.                     break;
  50.                 }
  52.                 long time = System.nanoTime();
  53.                 if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
  54.                     // timeoutMillis elapsed without anything selected.
  55.                     selectCnt = 1;
  56.                 } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
  57.                         selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
  58.                     // The selector returned prematurely many times in a row.
  59.                     // Rebuild the selector to work around the problem.
  60.                     logger.warn(
  61.                             "Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
  62.                             selectCnt, selector);
  64.                     rebuildSelector();
  65.                     selector = this.selector;
  67.                     // Select again to populate selectedKeys.
  68.                     selector.selectNow();
  69.                     selectCnt = 1;
  70.                     break;
  71.                 }
  73.                 currentTimeNanos = time;
  74.             }
  76.             if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {
  77.                 if (logger.isDebugEnabled()) {
  78.                     logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
  79.                             selectCnt - 1, selector);
  80.                 }
  81.             }
  82.         } catch (CancelledKeyException e) {
  83.             if (logger.isDebugEnabled()) {
  84.                 logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
  85.                         selector, e);
  86.             }
  87.             // Harmless exception - log anyway
  88.         }
  89.     }

在这种情况下，每次对selectCnt这个标志位进行自增的操作，后续通过计算:

  1. time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos

若满足，则seletCnt重新置为1，最后若一旦超过SELECTOR_AUTO_REBUILD_THRESHOLD(512)，那么需要重建selector，Netty正是通过这种方式规避了空轮询的bug。

发表评论

表情：
评论列表 （有 0 条评论，141人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Netty源码阅读之编码器简析

        上回主要聊了一下Netty中的解码器，那么既然有解码，也必须得聊下编码过程了，下面将对Netty中的编码器作一下总结：   1.编码器简介 作为解码的逆过程，编码

    悠悠悠悠/ 0 赞/ 59 阅读

    相关 netty源码阅读之NioEventLoop

    初始阅读源码的时候，晦涩难懂，枯燥无味，一段时间之后就会觉得豁然开朗，被源码的魅力深深折服。 接下去要阅读的是netty的一个重要组件，NioEventLoop。 将会分为

    你的名字你的名字/ 0 赞/ 329 阅读