netty的启动流程分析

发表于 | 分类于 | 评论数： | 阅读次数：

netty 是一个基于异步事件驱动实现的网络编程框架，它的内部使用了大量的异步编程方法，这是它性能高效的一个原因，但同时也使得代码阅读起来更加困难，本文就尝试分析下它的启动过程

按照惯例先上源码，这一段是 netty server 的启动代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

try {
    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(eventLoopGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new EchoServerHandler());
            }
        });

    ChannelFuture future = bootstrap.bind(port).sync();
    future.channel().closeFuture().sync();
} finally {
    eventLoopGroup.shutdownGracefully();
}

对应的 Handler 代码如下

1
2
3
4
5
6
7
8
9
10
11
12
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.write(msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

我们可以通过调试一步步跟踪代码的执行流程，由于以下原因，调试的流程可能不那么流畅

  • 注册 FutureListener，在任务执行完后通过回调执行相关逻辑
  • 提交任务到 EventLoop，将任务放到 EventLoop 线程中执行

为方便跟踪代码，特将整个启动流程的执行顺序梳理出来，进而整理成下图所示的流程图

首先看下 initAndRegister 的执行流程，它负责创建和初始化 channel，并将其注册到 EventLoop 上，同时将原生的 java channel 注册到 selector 上。

newChannel 和 initChannel

newChannel

netty 首先使用 channelFactory.newChannel 创建一个 channel 的实例，它的类型是在配置 ServerBootstrap 时指定的 .channel(NioServerSocketChannel.class)

很显然要通过反射创建的实例

1
2
3
4
5
6
7
public T newChannel() {
    try {
        return constructor.newInstance();
    } catch (Throwable t) {
        throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
    }
}

initChannel

在创建好 channel 实例之后，先要设置它的属性，这些都可以通过 ServerBootstrap 进行配置

1
2
setChannelOptions(channel, newOptionsArray(), logger);
setAttributes(channel, attrs0().entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY));

然后获取 channel 的 pipeline，注意每个 channel 会对应一个 pipeline。拿到 pipeline 之后会调用其 addLast 方法注册一个 ChannelInitializer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) {
        final ChannelPipeline pipeline = ch.pipeline();
        ChannelHandler handler = config.handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});

跟踪其调用流程，最终会执行如下代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// DefaultChannelPipeline#callHandlerCallbackLater
// ctx为包裹ChannelInitizlizer的ChannelHandlerContext
// added = true
private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
    assert !registered;

    PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
    PendingHandlerCallback pending = pendingHandlerCallbackHead;
    if (pending == null) {
        // 将head设置为PendingHandlerAddedTask
        pendingHandlerCallbackHead = task;
    } else {
        // Find the tail of the linked-list.
        while (pending.next != null) {
            pending = pending.next;
        }
        pending.next = task;
    }
}

此时 pendingHandlerCallbackHead 链表为空，并且调用该方法时入参 add 为 true，因此会将 head 设置为 PendingHandlerAddedTask。

我们来看下 PendingHandlerAddedTask 的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
private final class PendingHandlerAddedTask extends PendingHandlerCallback {

    PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) {
        super(ctx);
    }

    @Override
    public void run() {
        callHandlerAdded0(ctx);
    }

    @Override
    void execute() {
        EventExecutor executor = ctx.executor();
        if (executor.inEventLoop()) {
            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);
                }
                atomicRemoveFromHandlerList(ctx);
                ctx.setRemoved();
            }
        }
    }
}

它是 PendingHandlerCallback 的子类，它的职责是调用 callHandlerAdded0 方法触发 handler 的添加事件。

因为此时并不会触发其执行，所以暂且不去看内部逻辑，待后面调用到时再做分析

register

Channel 初始化好之后，接下来要将其注册到 EventLoop 上

1
config().group().register(channel)

config().group() 用于获取 ServerBootstrap 配置的 EventLoopGroup，拿到 EventLoopGroup 之后再调用其 register 方法

MultithreadEventLoopGroup.register 方法实现如下

1
return next().register(channel);

它先获取一个 EventLoop，然后调用其 register 方法。继续跟踪到 SingleThreadEventLoop 的 register 方法实现

1
promise.channel().unsafe().register(this, promise);

它是先获取 channel 的内部属性 unsafe，然后调用其 register 方法，如下是其核心代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
if (eventLoop.inEventLoop()) {
    register0(promise);
} else {
    try {
        // 这里
        eventLoop.execute(new Runnable() {
            @Override
            public void run() {
                // 任务一
                register0(promise);
            }
        });
    } catch (Throwable t) {
        // 忽略
    }
}

因为此时是在主线程中执行，所以会执行到第 6 行，往 EventLoop 中提交一个任务，用于完成 register0 操作，可以将其记做任务一

register0 需要等到 EventLoop 调度后执行。

那么 EventLoop 是在什么时候启动的呢？这需要跟进到 eventLoop.execute 内部

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private void execute(Runnable task, boolean immediate) {
    // 此时在主线程中执行，因此为false
    boolean inEventLoop = inEventLoop();
    // 添加到任务队列
    addTask(task);
    if (!inEventLoop) {
        // 启动线程
        startThread();
        // 忽略
    }

    if (!addTaskWakesUp && immediate) {
        wakeup(inEventLoop);
    }
}

因为此时是在主线程中执行，而不是 EventLoop 线程，所以除了会将任务添加到执行队列外，还会启动线程，之后 EventLoop 就进入到循环中，开始处理任务

regFuture.addListener

回到 doBind，在 initAndRegister 执行完之后，因为是异步操作 regFuture 还未完成，所以会调用 regFuture.addListener 注册一个 FutureListener ，以便等到注册完成之后执行 doBind0 操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        Throwable cause = future.cause();
        if (cause != null) {
            // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
            // IllegalStateException once we try to access the EventLoop of the Channel.
            promise.setFailure(cause);
        } else {
            // Registration was successful, so set the correct executor to use.
            // See https://github.com/netty/netty/issues/2586
            promise.registered();

            doBind0(regFuture, channel, localAddress, promise);
        }
    }
});

register0

再回到 register，这次是因为 EventLoop 已经开始执行，取到了 任务一 并执行，也就是说开始执行 register0 操作了

doRegister

将内部的 java channel 注册到 selector 上

1
2
3
4
5
6
7
8
9
10
11
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            // 忽略
        }
    }
}

注意第五行的 this 是 netty 的 channel，对于本文的场景实际是 NioServerSocketChannel。这样做的目的是将其作为 attachment 存储在 selectionKey 上，方便后面使用

pipeline.invokeHandlerAddedIfNeeded

它的调用链路比较长，简单来讲就是执行 pipeline 上所有的 PendingHandlerCallback，对于本文的场景就是执行 initChannel 时添加的 PendingHandlerAddedTask

DefaultChannelPipeline.callHandlerAdded0 -> AbstractChannelHandlerContext.callHandlerAdded -> ChannelInitializer.handlerAdded -> ChannelInitializer.initChannel

可以看到最终会执行 ChannelInitializer.initChannel 方法，那么这个 ChannelInitializer 是什么呢？

它实际上是在 initChannel 时添加的，再回顾下代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) {
        final ChannelPipeline pipeline = ch.pipeline();
        ChannelHandler handler = config.handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});

它的作用是将用户配置的 handler 添加到 pipeline，并且往 EventLoop 提交一个任务，这个任务是往 pipeline 添加 ServerBootstrapAcceptor，我们把这个任务记做任务二

safeSetSuccess

更新 future 的状态为成功，同时通知所有的 listener。这里只有一个 listener，它是在 regFuture.addListener 执行时添加的，再次回顾下代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        Throwable cause = future.cause();
        if (cause != null) {
            // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
            // IllegalStateException once we try to access the EventLoop of the Channel.
            promise.setFailure(cause);
        } else {
            // Registration was successful, so set the correct executor to use.
            // See https://github.com/netty/netty/issues/2586
            promise.registered();

            doBind0(regFuture, channel, localAddress, promise);
        }
    }
});

很明显会触发 doBind0 的执行

doBind0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {

    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    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());
            }
        }
    });
}

它是往 EventLoop 中提交了一个任务，通过 EventLoop 执行 channel.bind 操作。我们把这个任务记做任务三

执行任务二

EventLoop 取出任务二执行，将 ServerBootstrapAcceptor 添加到 pipeline 上

1
pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));

执行任务三

EventLoop 取出任务三并执行

1
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

bind 操作通过 pipeline 进行传递，从 tail 传递到 head，最终通过 head 调用 unsafe.bind 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
boolean wasActive = isActive();
try {
    doBind(localAddress);
} catch (Throwable t) {
    safeSetFailure(promise, t);
    closeIfClosed();
    return;
}

if (!wasActive && isActive()) {
    invokeLater(new Runnable() {
        @Override
        public void run() {
            pipeline.fireChannelActive();
        }
    });
}

safeSetSuccess(promise);

doBind 依赖于具体的 channel 实现，这里是 NioServerSocketChannel ，直接调用 java channel 的 bind 方法

1
2
3
4
5
6
7
protected void doBind(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        javaChannel().bind(localAddress, config.getBacklog());
    } else {
        javaChannel().socket().bind(localAddress, config.getBacklog());
    }
}

safeSetSuccess 修改 future 状态为成功，通知所有的 listener bind 事件已执行完

0%