1
@SuppressWarnings("unchecked")
2
@Override
3
protected void doRegister(ChannelPromise promise) {
4
  assert registration == null;
5
  // IoHandle 是 (AbstractNioUnsafe) unsafe()
6
  // IoHandle 获取 Channel 关联到 selector 返回 SelectionKey
7
  // 通过 IoRegistration 关联 IoHandle 和 SelectionKey，返回 IoRegistration 对象
8
  ((IoEventLoop) eventLoop()).register((AbstractNioUnsafe) unsafe()).addListener(f -> {
9
    if (f.isSuccess()) {
10
      // Channel 在回调中关联 IoRegistration
11
      registration = (IoRegistration) f.getNow();
12
      // 设置 registerPromise 状态，通知监听器
13
      promise.setSuccess();
14
    } else {
15
      promise.setFailure(f.cause());
16
    }
17
  });
18
}

1
@Override
2
public final Future<IoRegistration> register(final IoHandle handle) {
3
    Promise<IoRegistration> promise = newPromise();
4
    if (inEventLoop()) {
5
        registerForIo0(handle, promise);
6
    } else {
7
        execute(() -> registerForIo0(handle, promise));
8
    }
9

10
    return promise;
11
}
12

13
private void registerForIo0(final IoHandle handle, Promise<IoRegistration> promise) {
14
    assert inEventLoop();
15
    final IoRegistration registration;
16
    try {
17
        // 通过 ioHandler 返回 IoRegistration
18
        registration = ioHandler.register(handle);
19
    } catch (Exception e) {
20
        promise.setFailure(e);
21
        return;
22
    }
23
    numRegistrations.incrementAndGet();
24
    // 设置 promise 状态，回调里返回 IoRegistration
25
    promise.setSuccess(new IoRegistrationWrapper(registration));
26
}
27

28
@Override
29
public IoRegistration register(IoHandle handle)
30
    throws Exception {
31
  NioIoHandle nioHandle = nioHandle(handle);
32
  NioIoOps ops = NioIoOps.NONE;
33
  boolean selected = false;
34
  for (;;) {
35
    try {
36
      // 创建 IORegistration
37
      return new DefaultNioRegistration(executor, nioHandle, ops, unwrappedSelector());
38
    } catch (CancelledKeyException e) {
39
      if (!selected) {
40
        // Force the Selector to select now as the "canceled" SelectionKey may still be
41
        // cached and not removed because no Select.select(..) operation was called yet.
42
        selectNow();
43
        selected = true;
44
      } else {
45
        // We forced a select operation on the selector before but the SelectionKey is still cached
46
        // for whatever reason. JDK bug ?
47
        throw e;
48
      }
49
    }
50
  }
51
}
52

53
final class DefaultNioRegistration implements IoRegistration {
54
  private final NioIoHandle handle;
55
  private volatile SelectionKey key;
56

57
  DefaultNioRegistration(ThreadAwareExecutor executor, NioIoHandle handle, NioIoOps initialOps, Selector selector)
58
      throws IOException {
59
    this.handle = handle;
60
    // IoHandle 获取 channel 关联到 selector，返回 SelectionKey
61
    // 注意最后一个参数 this，把自己当参数传进去
62
    key = handle.selectableChannel().register(selector, initialOps.value, this);
63
  }
64
}
65

66
public final SelectionKey register(Selector sel, int ops, Object att)
67
    throws ClosedChannelException
68
{
69
    if ((ops & ~validOps()) != 0)
70
        throw new IllegalArgumentException();
71
    if (!isOpen())
72
        throw new ClosedChannelException();
73
    synchronized (regLock) {
74
        if (isBlocking())
75
            throw new IllegalBlockingModeException();
76
        synchronized (keyLock) {
77
            // re-check if channel has been closed
78
            if (!isOpen())
79
                throw new ClosedChannelException();
80
            // 判断是否重复注册
81
            SelectionKey k = findKey(sel);
82
            if (k != null) {
83
                k.attach(att);
84
                k.interestOps(ops);
85
            } else {
86
                // New registration
87
                // 向 selector 注册操作数，返回一个 key
88
                // att 是 IoRegistration，会绑定到 key 的 attach 里
89
                k = ((AbstractSelector)sel).register(this, ops, att);
90
                // 把 key 添加到 Channel 的 SelectionKey[] 中
91
                addKey(k);
92
            }
93
            return k;
94
        }
95
    }
96
}

1
((IoEventLoop) eventLoop()).register((AbstractNioUnsafe) unsafe()).addListener(f -> {
2
  if (f.isSuccess()) {
3
    // Channel 在回调中关联 IoRegistration
4
    registration = (IoRegistration) f.getNow();
5
    // 设置 registerPromise 状态，通知 Listener
6
    promise.setSuccess();
7
  } else {
8
    promise.setFailure(f.cause());
9
  }
10
});

1
registerPromise.addListener(new ChannelFutureListener() {
2
  @Override
3
  public void operationComplete(ChannelFuture future) throws Exception {
4
    // ...
5

6
    // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
7
    // user may already fire events through the pipeline in the ChannelFutureListener.
8
    // 执行 ChannelInitializer 的 config.handler() 添加用户配置的 handler，最后删除当前 ChannelInitializer
9
    pipeline.invokeHandlerAddedIfNeeded();
10
    // 设置状态，通知 promise 的 Listener
11
    safeSetSuccess(promise);
12
    // 递归调用 ChannelHandlerContext 所有的 channelRegistered 方法（事件传递）
13
    pipeline.fireChannelRegistered();
14
    // ...
15
  }
16
});

1
@Override
2
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
3
  if (ctx.channel().isRegistered()) {
4
    // This should always be true with our current DefaultChannelPipeline implementation.
5
    // The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering
6
    // surprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers
7
    // will be added in the expected order.
8
    if (initChannel(ctx)) {
9
      // We are done with init the Channel, removing the initializer now.
10
      // 源码注解已经说明，初始化后，移除 ChannelInitializer
11
      // 这里移除的是 ChannelInitializer 里 initMap 的 ctx
12
      removeState(ctx);
13
    }
14
  }
15
}
16

17
@SuppressWarnings("unchecked")
18
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
19
  if (initMap.add(ctx)) { // Guard against re-entrance.
20
    try {
21
      // 执行 ChannelInitializer 的 initChannel 方法
22
      initChannel((C) ctx.channel());
23
    } catch (Throwable cause) {
24
      // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
25
      // We do so to prevent multiple calls to initChannel(...).
26
      exceptionCaught(ctx, cause);
27
    } finally {
28
      if (!ctx.isRemoved()) {
29
        // 移除当前 ChannelInitializer
30
        ctx.pipeline().remove(this);
31
      }
32
    }
33
    return true;
34
  }
35
  return false;
36
}

1
final class ChannelHandlerMask {
2
    // Using to mask which methods must be called for a ChannelHandler.
3
    static final int MASK_EXCEPTION_CAUGHT = 1;
4
    static final int MASK_CHANNEL_REGISTERED = 1 << 1;
5
    static final int MASK_CHANNEL_UNREGISTERED = 1 << 2;
6
    static final int MASK_CHANNEL_ACTIVE = 1 << 3;
7
    static final int MASK_CHANNEL_INACTIVE = 1 << 4;
8
    static final int MASK_CHANNEL_READ = 1 << 5;
9
    static final int MASK_CHANNEL_READ_COMPLETE = 1 << 6;
10
    static final int MASK_USER_EVENT_TRIGGERED = 1 << 7;
11
    static final int MASK_CHANNEL_WRITABILITY_CHANGED = 1 << 8;
12
    static final int MASK_BIND = 1 << 9;
13
    static final int MASK_CONNECT = 1 << 10;
14
    static final int MASK_DISCONNECT = 1 << 11;
15
    static final int MASK_CLOSE = 1 << 12;
16
    static final int MASK_DEREGISTER = 1 << 13;
17
    static final int MASK_READ = 1 << 14;
18
    static final int MASK_WRITE = 1 << 15;
19
    static final int MASK_FLUSH = 1 << 16;
20

21
    // 事件集合
22
    static final int MASK_ONLY_INBOUND =  MASK_CHANNEL_REGISTERED |
23
            MASK_CHANNEL_UNREGISTERED | MASK_CHANNEL_ACTIVE | MASK_CHANNEL_INACTIVE | MASK_CHANNEL_READ |
24
            MASK_CHANNEL_READ_COMPLETE | MASK_USER_EVENT_TRIGGERED | MASK_CHANNEL_WRITABILITY_CHANGED;
25
    private static final int MASK_ALL_INBOUND = MASK_EXCEPTION_CAUGHT | MASK_ONLY_INBOUND;
26
    static final int MASK_ONLY_OUTBOUND =  MASK_BIND | MASK_CONNECT | MASK_DISCONNECT |
27
            MASK_CLOSE | MASK_DEREGISTER | MASK_READ | MASK_WRITE | MASK_FLUSH;
28
    private static final int MASK_ALL_OUTBOUND = MASK_EXCEPTION_CAUGHT | MASK_ONLY_OUTBOUND;
29
}
30

31
private static boolean skipContext(
32
    AbstractChannelHandlerContext ctx, EventExecutor currentExecutor, int mask, int onlyMask) {
33
  // Ensure we correctly handle MASK_EXCEPTION_CAUGHT which is not included in the MASK_EXCEPTION_CAUGHT
34
  // 通过位运算，筛选实现了某个事件方法的 HandlerContext
35
  return (ctx.executionMask & (onlyMask | mask)) == 0 ||
36
      // We can only skip if the EventExecutor is the same as otherwise we need to ensure we offload
37
      // everything to preserve ordering.
38
      //
39
      // See https://github.com/netty/netty/issues/10067
40
      (ctx.executor() == currentExecutor && (ctx.executionMask & mask) == 0);
41
}

1
MASK_A = 1 = 0000 0001
2
MASK_B = 1 << 1 = 0000 0010
3
MASK_C = 1 << 2 = 0000 0100
4

5
| 运算可以合并事件
6

7
  表示 A 事件和 B 事件的集合
8
  MASK_AB = MASK_A | MASK_B = 0000 0011
9

10
& 运算可以过滤事件
11

12
  表示 A 事件在 AB 事件集合中
13
  MASK_A & MASK_AB = 0000 0001
14

15
  表示 C 事件不在 AB 事件集合中
16
  MASK_C & MASK_AB = 0000 0000

1
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
2
  // invokeHandler() 防止 handler 还没添加完成，就开始接收事件
3
  if (invokeHandler()) {
4
    try {
5
      // DON'T CHANGE
6
      // Duplex handlers implements both out/in interfaces causing a scalability issue
7
      // see https://bugs.openjdk.org/browse/JDK-8180450
8
      final ChannelHandler handler = handler();
9
      // 获取第一个 HandlerContext
10
      final DefaultChannelPipeline.HeadContext headContext = pipeline.head;
11
      if (handler == headContext) {
12
        // 因为第一个 HandlerContext 是 netty 定义的，由它执行 unsafe.bind 比较安全
13
        // 第一个 HandlerContext 执行 bind 对端口进行绑定
14
        headContext.bind(this, localAddress, promise);
15
      } else if (handler instanceof ChannelDuplexHandler) {
16
        // 递归调用 bind 方法，继续找前一个 HandlerContext
17
        ((ChannelDuplexHandler) handler).bind(this, localAddress, promise);
18
      } else if (handler instanceof ChannelOutboundHandlerAdapter) {
19
        ((ChannelOutboundHandlerAdapter) handler).bind(this, localAddress, promise);
20
      } else {
21
        ((ChannelOutboundHandler) handler).bind(this, localAddress, promise);
22
      }
23
    } catch (Throwable t) {
24
      notifyOutboundHandlerException(t, promise);
25
    }
26
  } else {
27
    bind(localAddress, promise);
28
  }
29
}

1
@Override
2
public void bind(
3
        ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
4
    unsafe.bind(localAddress, promise);
5
}
6

7
@Override
8
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
9
    // ...
10
    boolean wasActive = isActive();
11
    try {
12
        doBind(localAddress);
13
    } catch (Throwable t) {
14
        safeSetFailure(promise, t);
15
        closeIfClosed();
16
        return;
17
    }
18

19
    if (!wasActive && isActive()) {
20
        // 递归执行 fireChannelActive 方法，激活 Channel 事件
21
        invokeLater(new Runnable() {
22
            @Override
23
            public void run() {
24
                pipeline.fireChannelActive();
25
            }
26
        });
27
    }
28

29
    safeSetSuccess(promise);
30
}
31

32
@Override
33
protected void doBind(SocketAddress localAddress) throws Exception {
34
    javaChannel().bind(localAddress, config.getBacklog());
35
}
36

37
@Override
38
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
39
    synchronized (stateLock) {
40
        ensureOpen();
41
        if (localAddress != null)
42
            throw new AlreadyBoundException();
43
        // 根据操作系统，选择调用
44
        if (isUnixSocket()) {
45
            localAddress = unixBind(local, backlog);
46
        } else {
47
            localAddress = netBind(local, backlog);
48
        }
49
    }
50
    return this;
51
}

1
@Override
2
public final ChannelPipeline fireChannelActive() {
3
    if (head.executor().inEventLoop()) {
4
        if (head.invokeHandler()) {
5
            head.channelActive(head);
6
        } else {
7
            head.fireChannelActive();
8
        }
9
    } else {
10
        head.executor().execute(this::fireChannelActive);
11
    }
12
    return this;
13
}
14

15
@Override
16
public void channelActive(ChannelHandlerContext ctx) {
17
    // 递归调用 fireChannelActive()
18
    ctx.fireChannelActive();
19
    // Channel 上开启读事件（OP_ACCEPT）的监听
20
    readIfIsAutoRead();
21
}

1
@Override
2
protected void doBeginRead() throws Exception {
3
    // Channel.read() or ChannelHandlerContext.read() was called
4
    IoRegistration registration = this.registration;
5
    if (registration == null || !registration.isValid()) {
6
        return;
7
    }
8
    // 设置 readPending
9
    readPending = true;
10
    // 注册 OP_ACCEPT 监听事件
11
    addAndSubmit(readOps);
12
}
13

14
protected void addAndSubmit(NioIoOps addOps) {
15
    // 获取当前 selector 注册的事件
16
    int interestOps = selectionKey().interestOps();
17
    // 如果没有当前的事件，注册该事件
18
    if (!addOps.isIncludedIn(interestOps)) {
19
        try {
20
            registration().submit(NioIoOps.valueOf(interestOps).with(addOps));
21
        } catch (Exception e) {
22
            throw new ChannelException(e);
23
        }
24
    }
25
}