2.91 3.注册channel

1. 要分析的代码

ChannelFuture regFuture = config().group().register(channel)

2. 选择一个NioEventLoop

• config()返回ServerBootstrap
• Serverbootstrap的group()返回EventLoopGroup，EventLoopGroup继承了MultithreadEventLoopGroup，所以调用的是MultithreadEventLoopGroup的register方法
• MultithreadEventLoopGroup#register

public ChannelFuture register(Channel channel) {
    return next().register(channel);
}

next()会调用到MultithreadEventExecutorGroup#next

public EventExecutor next() {
    //这个chooser要么是PowerOfTwoEventExecutorChooser，要么是GenericEventExecutorChooser
    return chooser.next();
}

chooser的next方法就是选择一个NioEventLoop选择一个NioEventLoop调用他的register方法 最终调用到SingleThreadEventLoop#register

3. 创建一个Future

• SingleThreadEventLoop#register

public ChannelFuture register(Channel channel) {
    return register(new DefaultChannelPromise(channel, this));
}

把Channel和EventLoopGroup传入DefaultChannelPromise构造，先看看DefaultChannelPromise

3.1. DefaultChannelFuture类体系

JDK的Future提供了获取异步任务结果的接口

Future
	cancel
	isCancelled
	isDone
	get
	get

而Netty的Future除了以上这些外，还提供了sync() 和 await() 用于阻塞等待， Listeners用于任务结束时回调【相比于JDK的Future的主动轮询，Netty的属于异步回调】 ChannelFuture继承了Netty的Future，则是将Channel【IO操作】和Future绑定在一起 Promise继承了Netty的Future，提供了setSucess和setFailure方法用于唤醒sync和await的线程 ChannelPromise则结合了ChannelFuture和Promise 最底层的的DefaultChannelPromise就是ChannelPromise默认的实现类

如上述我们有两种编程方式，

• 同步阻塞
  • 一种时sync/await阻塞等待，直至有线程调用setSuccess/setFailure通知
• 异步回调
  • 另一种时绑定listener实例，有结果时异步通知

4. 调用Unsafe进行register

将创建完的DefaultChannelPromise传入register

public ChannelFuture register(final ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    //AbstractUnsafe#register把当前channel关联到eventLoop里面
    promise.channel().unsafe().register(this, promise);
    return promise;
}

• AbstractUnsafe

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    if (eventLoop == null) {
        throw new NullPointerException("eventLoop");
    }
    if (isRegistered()) {
        promise.setFailure(new IllegalStateException("registered to an event loop already"));
        return;
    }
    if (!isCompatible(eventLoop)) {
        promise.setFailure(
                new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
        return;
    }

	//终于把NioEventLoop关联到channel里了
    AbstractChannel.this.eventLoop = eventLoop;

	//如果当前线程是在EventLoop中的，那么直接注册
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
    	//否则做成Runnable实例丢进SingleThreadEventExecutor#execute
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            logger.warn(
                    "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                    AbstractChannel.this, t);
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }
}

4.1. 把channel关联到NioEventLoop中

AbstractChannel.this.eventLoop = eventLoop;

4.2. 实际注册

private void register0(ChannelPromise promise) {
    try {
        // check if the channel is still open as it could be closed in the mean time when the register
        // call was outside of the eventLoop
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration = neverRegistered;
        //真的要注册了
        doRegister();
        neverRegistered = false;
        registered = true;


        //调用handler的handlerAdded方法
        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
    	//调用Handler的channelRegistered方法
        pipeline.fireChannelRegistered();
        // Only fire a channelActive if the channel has never been registered. This prevents firing
        // multiple channel actives if the channel is deregistered and re-registered.
        //这里会返回false，只有在bind的时候才会返回true
        if (isActive()) {
            if (firstRegistration) {
            	//所以不会调用handler的channelActive方法
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                // This channel was registered before and autoRead() is set. This means we need to begin read
                // again so that we process inbound data.
                //
                // See https://github.com/netty/netty/issues/4805
                beginRead();
            }
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

关键的两步，一是注册，二是调用handlerAdded方法

4.2.1. 将channel注册到jdk nio selector上

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
        	//调用jdk底层的channel注册
        	//eventLoop().unwrappedSelector()是jdk底层的selector
        	//0表示不关心任何事件
        	//this是当前的AbstractNioChannel作为jdk底层selector的attachment
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

4.2.2. 调用handlerAdded方法

final void invokeHandlerAddedIfNeeded() {
    assert channel.eventLoop().inEventLoop();
    if (firstRegistration) {
        firstRegistration = false;
        // We are now registered to the EventLoop. It's time to call the callbacks for the ChannelHandlers,
        // that were added before the registration was done.
        callHandlerAddedForAllHandlers();
    }
}


private void callHandlerAddedForAllHandlers() {
    final PendingHandlerCallback pendingHandlerCallbackHead;
    synchronized (this) {
        assert !registered;

        // This Channel itself was registered.
        registered = true;

        pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
        // Null out so it can be GC'ed.
        this.pendingHandlerCallbackHead = null;
    }

    // This must happen outside of the synchronized(...) block as otherwise handlerAdded(...) may be called while
    // holding the lock and so produce a deadlock if handlerAdded(...) will try to add another handler from outside
    // the EventLoop.
    PendingHandlerCallback task = pendingHandlerCallbackHead;
	//这里的task中其中有一个是io.netty.channel.DefaultChannelPipeline.PendingHandlerAddedTask#execute
    while (task != null) {
        task.execute();
        task = task.next;
    }
}

• DefaultChannelPipeline.PendingHandlerAddedTask#execute

void execute() {
        EventExecutor executor = ctx.executor();
        if (executor.inEventLoop()) {
        	//调用到io.netty.channel.DefaultChannelPipeline#callHandlerAdded0
            callHandlerAdded0(ctx);
        } else {
            try {
                executor.execute(this);
            } catch (RejectedExecutionException e) {
                if (logger.isWarnEnabled()) {
                    logger.warn(
                            "Can't invoke handlerAdded() as the EventExecutor {} rejected it, removing handler {}.",
                            executor, ctx.name(), e);
                }
                remove0(ctx);
                ctx.setRemoved();
            }
        }
    }
}

• DefaultChannelPipeline#callHandlerAdded0

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    try {
        // We must call setAddComplete before calling handlerAdded. Otherwise if the handlerAdded method generates
        // any pipeline events ctx.handler() will miss them because the state will not allow it.
        ctx.setAddComplete();
    	//调用io.netty.channel.ChannelInitializer#handlerAdded
    	//都是从pipeline开始传播
        ctx.handler().handlerAdded(ctx);
    } catch (Throwable t) {
        boolean removed = false;
        try {
            remove0(ctx);
            try {
                ctx.handler().handlerRemoved(ctx);
            } finally {
                ctx.setRemoved();
            }
            removed = true;
        } catch (Throwable t2) {
            if (logger.isWarnEnabled()) {
                logger.warn("Failed to remove a handler: " + ctx.name(), t2);
            }
        }

        if (removed) {
            fireExceptionCaught(new ChannelPipelineException(
                    ctx.handler().getClass().getName() +
                    ".handlerAdded() has thrown an exception; removed.", t));
        } else {
            fireExceptionCaught(new ChannelPipelineException(
                    ctx.handler().getClass().getName() +
                    ".handlerAdded() has thrown an exception; also failed to remove.", t));
        }
    }
}

ChannelInitializer#handlerAdded

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    if (ctx.channel().isRegistered()) {
        //调用io.netty.channel.ChannelInitializer#initChannel(io.netty.channel.ChannelHandlerContext)
        initChannel(ctx);
    }
}

4.2.2.1. 回调ChannelInitializer#initChannel

• ChannelInitializer#initChannel

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.
        try {
        	//终于回调到io.netty.channel.ChannelInitializer#initChannel
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
            // We do so to prevent multiple calls to initChannel(...).
            exceptionCaught(ctx, cause);
        } finally {
            remove(ctx);
        }
        return true;
    }
    return false;
}

ChannelInitializer是什么时候加进去的？ 见2.初始化channel.md