Netty 源码解析（八）: 回到 Channel 的 register 操作

private void register0(ChannelPromise promise) {     try {         ...         boolean firstRegistration = neverRegistered;         // *** 进行 JDK 底层的操作：Channel 注册到 Selector 上 ***         doRegister();
         neverRegistered = false;         registered = true;        // 到这里，就算是 registered 了
        // 这一步也很关键，因为这涉及到了 ChannelInitializer 的 init(channel)        // 我们之前说过，init 方法会将 ChannelInitializer 内部添加的 handlers 添加到 pipeline 中        pipeline.invokeHandlerAddedIfNeeded();
        // 设置当前 promise 的状态为 success        //   因为当前 register 方法是在 eventLoop 中的线程中执行的，需要通知提交 register 操作的线程        safeSetSuccess(promise);
        // 当前的 register 操作已经成功，该事件应该被 pipeline 上        //   所有关心 register 事件的 handler 感知到，往 pipeline 中扔一个事件        pipeline.fireChannelRegistered();
        // 这里 active 指的是 channel 已经打开        if (isActive()) {            // 如果该 channel 是第一次执行 register，那么 fire ChannelActive 事件            if (firstRegistration) {                pipeline.fireChannelActive();            } else if (config().isAutoRead()) {                // 该 channel 之前已经 register 过了，                // 这里让该 channel 立马去监听通道中的 OP_READ 事件                beginRead();            }        }    } catch (Throwable t) {        ...    }}

 @Override protected void doRegister() throws Exception {     boolean selected = false;     for (;;) {         try {             // 附 JDK 中 Channel 的 register 方法：             // public final SelectionKey register(Selector sel, int ops, Object att) {...}             selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);             return;        } catch (CancelledKeyException e) {            ...        }    }}

1pipeline.invokeHandlerAddedIfNeeded();

@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {    if (ctx.channel().isRegistered()) {        initChannel(ctx);    }}

 private boolean initChannel(ChannelHandlerContext ctx) throws Exception {     if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.         try {             // 1. 将把我们自定义的 handlers 添加到 pipeline 中             initChannel((C) ctx.channel());         } catch (Throwable cause) {             ...         } finally {             // 2. 将 ChannelInitializer 实例从 pipeline 中删除            remove(ctx);        }        return true;    }    return false;}

pipeline.fireChannelRegistered();

1@Override2public final ChannelPipeline fireChannelRegistered() {3    // 注意这里的传参是 head4    AbstractChannelHandlerContext.invokeChannelRegistered(head);5    return this;6}

 // next 此时是 head 2static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
     EventExecutor executor = next.executor();     // 执行 head 的 invokeChannelRegistered()     if (executor.inEventLoop()) {         next.invokeChannelRegistered();     } else {         executor.execute(new Runnable() {            @Override            public void run() {                next.invokeChannelRegistered();            }        });   }}

 private void invokeChannelRegistered() {     if (invokeHandler()) {         try {             // handler() 方法此时会返回 head             ((ChannelInboundHandler) handler()).channelRegistered(this);         } catch (Throwable t) {             notifyHandlerException(t);         }     } else {        fireChannelRegistered();    }}

@Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {    // 1. 这一步是 head 对于 channelRegistered 事件的处理。没有我们要关心的    invokeHandlerAddedIfNeeded();    // 2. 向后传播 Inbound 事件    ctx.fireChannelRegistered();}

1@Override2public ChannelHandlerContext fireChannelRegistered() {3    // 这里很关键4    // findContextInbound() 方法会沿着 pipeline 找到下一个 Inbound 类型的 handler5    invokeChannelRegistered(findContextInbound());6    return this;7}

 final ChannelFuture initAndRegister() {     Channel channel = null;     try {         channel = channelFactory.newChannel();         init(channel);     } catch (Throwable t) {         ...     }    // 我们上面说完了这行    ChannelFuture regFuture = config().group().register(channel);    // 如果在 register 的过程中，发生了错误    if (regFuture.cause() != null) {        if (channel.isRegistered()) {            channel.close();        } else {            channel.unsafe().closeForcibly();        }    }    // 源码中说得很清楚，如果到这里，说明后续可以进行 connect() 或 bind() 了，因为两种情况：    // 1. 如果 register 动作是在 eventLoop 中发起的，那么到这里的时候，register 一定已经完成    // 2. 如果 register 任务已经提交到 eventLoop 中，也就是进到了 eventLoop 中的 taskQueue 中，    //    由于后续的 connect 或 bind 也会进入到同一个 eventLoop 的 queue 中，所以一定是会先 register 成功，才会执行 connect 或 bind    return regFuture;}