1
private void doStartThread() {
2
  executor.execute(new Runnable() {
3
    @Override
4
    public void run() {
5
      // 获取锁
6
      processingLock.lock();
7
      // ...
8
      try {
9
        for (;;) {
10
          // 开启 bossGroup 线程 EventLoop 的主循环
11
          SingleThreadEventExecutor.this.run();
12
          success = true;
13

14
          int currentState = state;
15
          if (canSuspend(currentState)) {
16
              if (!STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this,
17
                      ST_SUSPENDING, ST_SUSPENDED)) {
18
                  // Try again as the CAS failed.
19
                  continue;
20
              }
21

22
              if (!canSuspend(ST_SUSPENDED) && STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this,
23
                          ST_SUSPENDED, ST_STARTED)) {
24
                  // Seems like there was something added to the task queue again in the meantime but we
25
                  // were able to re-engage this thread as the event loop thread.
26
                  continue;
27
              }
28
              suspend = true;
29
          }
30

31
          // 跳出循环
32
          break;
33
        }
34
      } catch (Throwable t) {
35
        unexpectedException = t;
36
        logger.warn("Unexpected exception from an event executor: ", t);
37
      } finally {
38
        // ...
39
      }
40
    }
41
  });
42
}

1
@Override
2
protected void run() {
3
  // 判断是否在 EventLoop 线程中
4
    assert inEventLoop();
5
    // 执行 ioHandler 的初始化（生命周期钩子函数）
6
    ioHandler.initialize();
7
    do {
8
    // 处理 I/O 就绪事件
9
        runIo();
10
        if (isShuttingDown()) {
11
          // ioHandler 销毁前执行（生命周期钩子函数）
12
            ioHandler.prepareToDestroy();
13
        }
14
        // Now run all tasks for the maximum configured amount of time before trying to run IO again.
15
        // 在规定时间内执行所有 EventLoop 的异步任务
16
        runAllTasks(maxTaskProcessingQuantumNs);
17

18
        // We should continue with our loop until we either confirmed a shutdown or we can suspend it.
19
        // 如源码注释所说，这里应该继续循环，直到我们认为可以停止
20
    } while (!confirmShutdown() && !canSuspend());
21
}
22

23
protected int runIo() {
24
    assert inEventLoop();
25
    // 通过 ioHandler 执行
26
    return ioHandler.run(context);
27
}
28

29
@Override
30
public int run(IoHandlerContext context) {
31
  int handled = 0;
32
  try {
33
    try {
34
      // 有任务调用一次 selectNow，及时处理 I/O 事件
35
      // 无任务，进入 SELECT 阻塞
36
      switch (selectStrategy.calculateStrategy(selectNowSupplier, !context.canBlock())) {
37
        case SelectStrategy.CONTINUE:
38
          return 0;
39

40
        case SelectStrategy.BUSY_WAIT:
41
          // fall-through to SELECT since the busy-wait is not supported with NIO
42

43
        case SelectStrategy.SELECT:
44
          // 通过 selector.select() 阻塞 I/O 监听事件
45
          select(context, wakenUp.getAndSet(false));
46

47
          // wakeUp 变量表示是否唤醒 selector
48
          // 每次新一轮循环，wakeUp 设置为 false
49
          // 当有任务执行时，会设置为 true 唤醒线程
50
          if (wakenUp.get()) {
51
            selector.wakeup();
52
          }
53
          // fall through
54
        default:
55
      }
56
    } catch (IOException e) {
57
      // If we receive an IOException here its because the Selector is messed up. Let's rebuild
58
      // the selector and retry. https://github.com/netty/netty/issues/8566
59
      rebuildSelector0();
60
      handleLoopException(e);
61
      return 0;
62
    }
63

64
    cancelledKeys = 0;
65
    needsToSelectAgain = false;
66
    // 处理 SelectedKey
67
    handled = processSelectedKeys();
68
  } catch (Error e) {
69
    throw e;
70
  } catch (Throwable t) {
71
    handleLoopException(t);
72
  }
73
  return handled;
74
}

1
private void select(IoHandlerContext runner, boolean oldWakenUp) throws IOException {
2
  Selector selector = this.selector;
3
  try {
4
    int selectCnt = 0;
5
    long currentTimeNanos = System.nanoTime();
6
    // 根据下一个任务执行时间判断 selector.select() 可以阻塞多久
7
    // 截止时间 = 当前时间 + 队列第一个任务剩余执行时间
8
    long selectDeadLineNanos = currentTimeNanos + runner.delayNanos(currentTimeNanos);
9

10
    for (;;) {
11
      // 计算超时，防止一直阻塞，每次循环最后会更新 currentTimeNanos
12
      // 阻塞时间 = 截止时间 - 当前时间 + 0.5毫秒（四舍五入）/ 1000000L（转毫秒）
13
      long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
14
      // 小于等于0，说明任务该执行了，跳出循环
15
      if (timeoutMillis <= 0) {
16
        if (selectCnt == 0) {
17
          selector.selectNow();
18
          selectCnt = 1;
19
        }
20
        break;
21
      }
22

23
      // If a task was submitted when wakenUp value was true, the task didn't get a chance to call
24
      // Selector#wakeup. So we need to check task queue again before executing select operation.
25
      // If we don't, the task might be pended until select operation was timed out.
26
      // It might be pended until idle timeout if IdleStateHandler existed in pipeline.
27
      if (!runner.canBlock() && wakenUp.compareAndSet(false, true)) {
28
        selector.selectNow();
29
        selectCnt = 1;
30
        break;
31
      }
32

33
      // 规定时间内执行 select
34
      int selectedKeys = selector.select(timeoutMillis);
35
      selectCnt ++;
36

37
      // 如果有事件、任务唤醒等操作，跳出循环
38
      if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || !runner.canBlock()) {
39
        // - Selected something,
40
        // - waken up by user, or
41
        // - the task queue has a pending task.
42
        // - a scheduled task is ready for processing
43
        break;
44
      }
45
      if (Thread.interrupted()) {
46
        // Thread was interrupted so reset selected keys and break so we not run into a busy loop.
47
        // As this is most likely a bug in the handler of the user or it's client library we will
48
        // also log it.
49
        //
50
        // See https://github.com/netty/netty/issues/2426
51
        if (logger.isDebugEnabled()) {
52
          logger.debug("Selector.select() returned prematurely because " +
53
              "Thread.currentThread().interrupt() was called. Use " +
54
              "NioHandler.shutdownGracefully() to shutdown the NioHandler.");
55
        }
56
        selectCnt = 1;
57
        break;
58
      }
59

60
      long time = System.nanoTime();
61
      // 当前时间 - 阻塞时间 >= 循环开始时间
62
      // 等同于 阻塞时间 >= select 执行时间（当前时间 - 循环开始时间）
63
      if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
64
        // timeoutMillis elapsed without anything selected.
65
        // 正常超时 selectCnt 重置为 1
66
        selectCnt = 1;
67
        // 超出阈值，重建 selector
68
      } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
69
          selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
70
        // The code exists in an extra method to ensure the method is not too big to inline as this
71
        // branch is not very likely to get hit very frequently.
72
        selector = selectRebuildSelector(selectCnt);
73
        selectCnt = 1;
74
        break;
75
      }
76
      // 更新当前时间
77
      currentTimeNanos = time;
78
    }
79
    //...
80
  } catch (CancelledKeyException e) {
81
    if (logger.isDebugEnabled()) {
82
      logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
83
          selector, e);
84
    }
85
    // Harmless exception - log anyway
86
  }
87
}

1
void rebuildSelector0() {
2
    final Selector oldSelector = selector;
3
    final SelectorTuple newSelectorTuple;
4

5
    if (oldSelector == null) {
6
        return;
7
    }
8

9
    try {
10
      // 创建新的 selector
11
        newSelectorTuple = openSelector();
12
    } catch (Exception e) {
13
        logger.warn("Failed to create a new Selector.", e);
14
        return;
15
    }
16

17
    // Register all channels to the new Selector.
18
    // 通过 IoRegistration 把旧的 channels 注册到新的 selector
19
    int nChannels = 0;
20
    for (SelectionKey key : oldSelector.keys()) {
21
        DefaultNioRegistration handle = (DefaultNioRegistration) key.attachment();
22
        try {
23
            if (!key.isValid() || key.channel().keyFor(newSelectorTuple.unwrappedSelector) != null) {
24
                continue;
25
            }
26

27
            handle.register(newSelectorTuple.unwrappedSelector);
28
            nChannels++;
29
        } catch (Exception e) {
30
            logger.warn("Failed to re-register a NioHandle to the new Selector.", e);
31
            handle.cancel();
32
        }
33
    }
34

35
    selector = newSelectorTuple.selector;
36
    unwrappedSelector = newSelectorTuple.unwrappedSelector;
37

38
    try {
39
        // time to close the old selector as everything else is registered to the new one
40
        oldSelector.close();
41
    } catch (Throwable t) {
42
        if (logger.isWarnEnabled()) {
43
            logger.warn("Failed to close the old Selector.", t);
44
        }
45
    }
46
  //...
47
}
48

49
private SelectorTuple openSelector() {
50
    final Selector unwrappedSelector;
51
    try {
52
      // 创建 JDK NIO 的 selector
53
        unwrappedSelector = provider.openSelector();
54
    } catch (IOException e) {
55
        throw new ChannelException("failed to open a new selector", e);
56
    }
57

58
    if (DISABLE_KEY_SET_OPTIMIZATION) {
59
        return new SelectorTuple(unwrappedSelector);
60
    }
61

62
    Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {
63
        @Override
64
        public Object run() {
65
            try {
66
                return Class.forName(
67
                        "sun.nio.ch.SelectorImpl",
68
                        false,
69
                        PlatformDependent.getSystemClassLoader());
70
            } catch (Throwable cause) {
71
                return cause;
72
            }
73
        }
74
    });
75

76
  //...
77

78
    final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
79
    final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
80
    // 反射替换 selectedKeySet 的数据结构，把 HashSet 替换成数组
81
    Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
82
        @Override
83
        public Object run() {
84
            try {
85
                Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
86
                Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
87

88
                if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) {
89
                    // Let us try to use sun.misc.Unsafe to replace the SelectionKeySet.
90
                    // This allows us to also do this in Java9+ without any extra flags.
91
                    long selectedKeysFieldOffset = PlatformDependent.objectFieldOffset(selectedKeysField);
92
                    long publicSelectedKeysFieldOffset =
93
                            PlatformDependent.objectFieldOffset(publicSelectedKeysField);
94

95
                    if (selectedKeysFieldOffset != -1 && publicSelectedKeysFieldOffset != -1) {
96
                        PlatformDependent.putObject(
97
                                unwrappedSelector, selectedKeysFieldOffset, selectedKeySet);
98
                        PlatformDependent.putObject(
99
                                unwrappedSelector, publicSelectedKeysFieldOffset, selectedKeySet);
100
                        return null;
101
                    }
102
                    // We could not retrieve the offset, lets try reflection as last-resort.
103
                }
104
                //...
105
                selectedKeysField.set(unwrappedSelector, selectedKeySet);
106
                publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
107
                return null;
108
            } catch (NoSuchFieldException | IllegalAccessException e) {
109
                return e;
110
            }
111
        }
112
    });
113
  //...
114
    selectedKeys = selectedKeySet;
115
}

1
public int run(IoHandlerContext context) {
2
  //...
3
  select(context, wakenUp.getAndSet(false));
4
  // 处理 SelectedKey
5
  handled = processSelectedKeys();
6
  //...
7
    return handled;
8
}
9

10
private int processSelectedKeys() {
11
    // selector 创建的时候通过反射复制 selectedKeys
12
    if (selectedKeys != null) {
13
        // 优化处理，使用数组
14
        return processSelectedKeysOptimized();
15
    } else {
16
        // 没有优化的处理，使用迭代器
17
        return processSelectedKeysPlain(selector.selectedKeys());
18
    }
19
}
20

21
private int processSelectedKeysOptimized() {
22
  int handled = 0;
23
  for (int i = 0; i < selectedKeys.size; ++i) {
24
    final SelectionKey k = selectedKeys.keys[i];
25
    // null out entry in the array to allow to have it GC'ed once the Channel close
26
    // See https://github.com/netty/netty/issues/2363
27
    // 有些 Channel 已经关闭，但 key 还存在，影响 GC
28
    selectedKeys.keys[i] = null;
29
    // 处理 key
30
    processSelectedKey(k);
31
    ++handled;
32

33
    if (needsToSelectAgain) {
34
      // null out entries in the array to allow to have it GC'ed once the Channel close
35
      // See https://github.com/netty/netty/issues/2363
36
      // 清除后面的 selectedKey，防止 Channel 关闭 key 没有遍历到
37
      selectedKeys.reset(i + 1);
38

39
      selectAgain();
40
      i = -1;
41
    }
42
  }
43
  return handled;
44
}
45

46
private void processSelectedKey(SelectionKey k) {
47
  // 从 SelectionKey 获取 IoRegistration
48
  final DefaultNioRegistration registration = (DefaultNioRegistration) k.attachment();
49
  if (!registration.isValid()) {
50
    try {
51
      registration.handle.close();
52
    } catch (Exception e) {
53
      logger.debug("Exception during closing " + registration.handle, e);
54
    }
55
    return;
56
  }
57
  // 从 SelectionKey 中获取操作数，通过 IoRegistration 的 handle 判断事件掩码
58
  // 选择对应的 Channel 执行
59
  registration.handle(k.readyOps());
60
}

1
ServerBootstrap b = new ServerBootstrap();
2
b.group(bossGroup, workerGroup)
3
 .channel(NioServerSocketChannel.class)
4
//...

1
@Override
2
public void handle(IoRegistration registration, IoEvent event) {
3
  try {
4
    NioIoEvent nioEvent = (NioIoEvent) event;
5
    // 获取当前事件的掩码
6
    NioIoOps nioReadyOps = nioEvent.ops();
7
    // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
8
    // the NIO JDK channel implementation may throw a NotYetConnectedException.
9
    if (nioReadyOps.contains(NioIoOps.CONNECT)) {
10
      // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
11
      // See https://github.com/netty/netty/issues/924
12
      removeAndSubmit(NioIoOps.CONNECT);
13

14
      unsafe().finishConnect();
15
    }
16

17
    // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
18
    if (nioReadyOps.contains(NioIoOps.WRITE)) {
19
      // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to
20
      // write
21
      forceFlush();
22
    }
23

24
    // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
25
    // to a spin loop
26
    // 通过位运算，判断当前事件掩码是否在 read 和 accept 事件集合中
27
    if (nioReadyOps.contains(NioIoOps.READ_AND_ACCEPT) || nioReadyOps.equals(NioIoOps.NONE)) {
28
      // 有两个实现，建立连接监听 OP_ACCPET 事件是 AbstractNioMessageChannel
29
      // 后续监听 OP_READ 事件是 AbstractNioByteChannel
30
      read();
31
    }
32
  } catch (CancelledKeyException ignored) {
33
    close(voidPromise());
34
  }
35
}

1
@Override
2
public void read() {
3
  assert eventLoop().inEventLoop();
4
  final ChannelConfig config = config();
5
  final ChannelPipeline pipeline = pipeline();
6
  // 获取内存大小预测器，分配缓冲区
7
  final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
8
  // 重置内存大小预测器的属性
9
  allocHandle.reset(config);
10

11
  boolean closed = false;
12
  Throwable exception = null;
13
  try {
14
    try {
15
      do {
16
        // 服务端这里会创建 channel 放到 readBuf 中
17
        int localRead = doReadMessages(readBuf);
18
        if (localRead == 0) {
19
          break;
20
        }
21
        if (localRead < 0) {
22
          closed = true;
23
          break;
24
        }
25
        // 增加已读信息数
26
        allocHandle.incMessagesRead(localRead);
27
      } while (continueReading(allocHandle));
28
    } catch (Throwable t) {
29
      exception = t;
30
    }
31

32
    int size = readBuf.size();
33
    for (int i = 0; i < size; i ++) {
34
      // 是否等待下一次读事件
35
      readPending = false;
36
      // 传递 read 事件
37
      pipeline.fireChannelRead(readBuf.get(i));
38
    }
39
    readBuf.clear();
40
    // 读结束，根据这次读取的字节数，动态调整下次的预测
41
    allocHandle.readComplete();
42
    // 传递 readComplete 事件
43
    pipeline.fireChannelReadComplete();
44

45
    if (exception != null) {
46
      closed = closeOnReadError(exception);
47

48
      pipeline.fireExceptionCaught(exception);
49
    }
50

51
    if (closed) {
52
      inputShutdown = true;
53
      if (isOpen()) {
54
        close(voidPromise());
55
      }
56
    }
57
  } finally {
58
    if (!readPending && !config.isAutoRead()) {
59
      removeReadOp();
60
    }
61
  }
62
}
63

64
@Override
65
protected int doReadMessages(List<Object> buf) throws Exception {
66
  // 使用 JDK 创建一个 socket 连接
67
    SocketChannel ch = SocketUtils.accept(javaChannel());
68

69
    try {
70
        if (ch != null) {
71
            // 将 socket 连接包装为 channel
72
            buf.add(new NioSocketChannel(this, ch));
73
            return 1;
74
        }
75
    } catch (Throwable t) {
76
        logger.warn("Failed to create a new channel from an accepted socket.", t);
77

78
        try {
79
            ch.close();
80
        } catch (Throwable t2) {
81
            logger.warn("Failed to close a socket.", t2);
82
        }
83
    }
84

85
    return 0;
86
}
87

88
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
89
  // 指定 OP_READ 操作数
90
    super(parent, ch, SelectionKey.OP_READ);
91
}

1
@SuppressWarnings("unchecked")
2
public void channelRead(ChannelHandlerContext ctx, Object msg) {
3
  final Channel child = (Channel) msg;
4
  // 添加 childHandler
5
  child.pipeline().addLast(childHandler);
6

7
  setChannelOptions(child, childOptions, logger);
8
  setAttributes(child, childAttrs);
9

10
  //...
11

12
  try {
13
    // 选择一个 worker 线程注册 channel，并且添加一个关闭的监听器
14
    childGroup.register(child).addListener(new ChannelFutureListener() {
15
      @Override
16
      public void operationComplete(ChannelFuture future) throws Exception {
17
        if (!future.isSuccess()) {
18
          forceClose(child, future.cause());
19
        }
20
      }
21
    });
22
  } catch (Throwable t) {
23
    forceClose(child, t);
24
  }
25
}
26

27
@Override
28
public ChannelFuture register(Channel channel) {
29
    return next().register(channel);
30
}
31

32
@Override
33
public EventExecutor next() {
34
  // 通过轮询的负载均衡算法选择一个 worker 线程
35
    return executors[idx.getAndIncrement() & executors.length - 1];
36
}

1
private void register0(ChannelPromise promise) {
2
  //...
3
  // 初始化 ChildHandler 的 ChannelInitializer
4
  pipeline.invokeHandlerAddedIfNeeded();
5
  // 设置状态，通知 promise 的 Listener
6
  safeSetSuccess(promise);
7
  // 递归调用 ChannelHandlerContext 所有的 channelRegistered 方法（事件传递）
8
  pipeline.fireChannelRegistered();
9

10
  if (isActive()) {
11
    if (firstRegistration) {
12
      pipeline.fireChannelActive();
13
    } else if (config().isAutoRead()) {
14
      beginRead();
15
    }
16
  }
17
  //...
18

19
  // selector 注册到 IoHandle 然后通过 IoRegistration 注册 IoHandle 到 IoHandler
20
  doRegister(registerPromise);
21
}

1
protected void doBeginRead() throws Exception {
2
  // Channel.read() or ChannelHandlerContext.read() was called
3
  IoRegistration registration = this.registration;
4
  if (registration == null || !registration.isValid()) {
5
    return;
6
  }
7

8
  readPending = true;
9

10
  addAndSubmit(readOps);
11
}
12

13
protected void addAndSubmit(NioIoOps addOps) {
14
    int interestOps = selectionKey().interestOps();
15
    if (!addOps.isIncludedIn(interestOps)) {
16
        try {
17
          // 向 selector 注册 OP_READ 操作数
18
            registration().submit(NioIoOps.valueOf(interestOps).with(addOps));
19
        } catch (Exception e) {
20
            throw new ChannelException(e);
21
        }
22
    }
23
}