【Netty源码分析】02 Netty Server 启动流程下

Reactor2020

发布于 2023-03-22 18:59:35

3030

发布于 2023-03-22 18:59:35

文章被收录于专栏：【云原生 • Prometheus】

上面我们分析了initAndRegister()方法的核心流程，Channel准备工作基本也都完成了：

Channel和NioEventLoop进行了关联；
Channel也注册到Selector上了；
NioEventLoop线程也启动完成，开始轮询事件、处理事件。

这里还遗漏了两个事情：channel和端口绑定以及channel向Selector注册OP_ACCEPT。这就是在doBind()方法中另一个重要的方法：doBind0()中进行完成的。

doBind0

if (regFuture.isDone()) {
    ChannelPromise promise = channel.newPromise();
    doBind0(regFuture, channel, localAddress, promise);
    return promise;
} else {
    final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
    //register还未完成，则添加listener，待注册完成再执行doBind0()进行server端口绑定
    regFuture.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            Throwable cause = future.cause();
            if (cause != null) {
                promise.setFailure(cause);
            } else {
                promise.registered();
                doBind0(regFuture, channel, localAddress, promise);
            }
        }
    });
    return promise;
}

上面代码一大堆，核心就是调用doBind0()方法，但是执行该方法前必须保证上一步initAndRegister()方法中执行完成。通过regFuture.isDone()进行判断，具体设置位置见下：

//AbstractChannel.AbstractUnsafe#register0
pipeline.invokeHandlerAddedIfNeeded();
// 将指定的promise标记为成功：regFuture.isDone()=true，doBind0()才能开始执行
safeSetSuccess(promise);
pipeline.fireChannelRegistered();

进行向下跟踪，来到了如下代码处，会发现需要调用channel.bind()方法，但是不是在当前线程中直接调用，而是封装成task放入到NioEventLoop的任务队列taskQueue中，由NioEventLoop线程执行：

private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {

    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
   });
}

这时的NioEventLoop线程是已经启动并开始工作的，所以channel.bind()这里是可以执行的。

层层调用最终是在pipeline中的head这个节点进行处理的：

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    assertEventLoop();

    if (!promise.setUncancellable() || !ensureOpen(promise)) {
        return;
    }

    //还没有绑定端口，isActive()返回false
    boolean wasActive = isActive();
    try {
        //调用底层java api，将channel绑定到具体端口上
        doBind(localAddress);
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }
    //经过上面绑定端口，这时isActive()=true
    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                //触发server handler的channelActive()方法
                pipeline.fireChannelActive();
            }
        });
    }
    safeSetSuccess(promise);
}

这个方法主要完成2件事：

doBind()：调用java api，将channel绑定到具体端口上；
pipeline.fireChannelActive()：将pipeline.fireChannelActive()放入到NioEventLoop线程中执行；

下面我们再来看下pipeline.fireChannelActive()：

public void channelActive(ChannelHandlerContext ctx) {
 ctx.fireChannelActive();
 readIfIsAutoRead();
}

该方法主要做2件事：

ctx.fireChannelActive()：触发handler#channelActive()调用，表示当前channel已处于激活状态，可以正常工作了；
readIfIsAutoRead()：从名称看就是，如果配置autoRead，调用readIfIsAutoRead()直接进行read操作；readIfIsAutoRead()会调用tail.read()，然后一层层往前查找，最终调用的是head#read()方法。

protected void doBeginRead() throws Exception {
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
        return;
    }

    readPending = true;

    final int interestOps = selectionKey.interestOps();
    // 将SelectionKey当前的操作位与注册操作位进行按位与操作，如果等于0，说明目前并没有设置注册操作位
    if ((interestOps & readInterestOp) == 0) {
        // Server Channel会在这里注册真正的ACCEPT事件
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

channel绑定好端口后，触发了channelActive()方法回调，channel真正进入可以正常工作状态，这时还差最后一步：注册OP_ACCEPT事件。