2.74 4.向selector注册读事件

1. 要分析的代码

• io.netty.channel.AbstractChannel.AbstractUnsafe#register中的register操作

//...
register0(promise);//...
//...

2. 打断点使用nc连接

其实跟服务器启动的时候向selector注册accept一样，我们看io.netty.channel.AbstractChannel.AbstractUnsafe#register0 使用nc访问，最终会进入pipeline.fireChannelActive()

private void register0(ChannelPromise promise) {
    try {
       //.....
       //在这行打个断点
        if (isActive()) {
            if (firstRegistration) {
                //调用pipeline的fireChannelActive
                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);
    }
}

3. 通过pipeline传播channelActive事件

• DefaultChannelPipeline#fireChannelActive

public final ChannelPipeline fireChannelActive() {
	//io.netty.channel.AbstractChannelHandlerContext#invokeChannelActive(io.netty.channel.AbstractChannelHandlerContext)
	//通过AbstractChannelHandlerContext传播
    AbstractChannelHandlerContext.invokeChannelActive(head);
    return this;
}

3.1. 经由AbstractChannelHandlerContext传播

AbstractChannelHandlerContext#invokeChannelActive

static void invokeChannelActive(final AbstractChannelHandlerContext next) {
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
    	//进入这里
    	//io.netty.channel.AbstractChannelHandlerContext#invokeChannelActive()
        next.invokeChannelActive();
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelActive();
            }
        });
    }
}

• AbstractChannelHandlerContext#invokeChannelActive()

private void invokeChannelActive() {
    if (invokeHandler()) {
        try {
        	//io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive
            ((ChannelInboundHandler) handler()).channelActive(this);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelActive();
    }
}

4. 先到达HeadContext

• DefaultChannelPipeline.HeadContext#channelActive

public void channelActive(ChannelHandlerContext ctx) throws Exception {
	ctx.fireChannelActive();
	//io.netty.channel.DefaultChannelPipeline.HeadContext#readIfIsAutoRead
	readIfIsAutoRead();
}

4.1. 调用Channel.read方法

• DefaultChannelPipeline.HeadContext#readIfIsAutoRead

private void readIfIsAutoRead() {
    if (channel.config().isAutoRead()) {
    	//io.netty.channel.AbstractChannel#read
        channel.read();
    }
}

• AbstractChannel#read

public Channel read() {
	//pipeline的tail开始传播到head，最后到io.netty.channel.AbstractChannel.AbstractUnsafe#beginRead
    pipeline.read();
    return this;
}

5. 由AbstractUnsafe进行读取操作

• AbstractChannel.AbstractUnsafe#beginRead

public final void beginRead() {
    assertEventLoop();

    if (!isActive()) {
        return;
    }

    try {
    	//io.netty.channel.nio.AbstractNioChannel#doBeginRead
        doBeginRead();
    } catch (final Exception e) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                pipeline.fireExceptionCaught(e);
            }
        });
        close(voidPromise());
    }
}

5.1. 注册读事件

• AbstractNioChannel#doBeginRead

protected void doBeginRead() throws Exception {
    // Channel.read() or ChannelHandlerContext.read() was called
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
        return;
    }

    readPending = true;
	//注册读事件
    final int interestOps = selectionKey.interestOps();
    if ((interestOps & readInterestOp) == 0) {
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}