Netty源码解读（二）-服务端源码讲解

简单Echo案例

代码是netty的源码，我添加了自己理解的中文注释。

了解了Netty的线程模型和组件之后，我们先看看如何写一个简单的Echo案例，后续的源码讲解都基于此案例。以下是服务端的代码：

public final class MyEchoServer {

    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final MyEchoServerHandler serverHandler = new MyEchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    // 说明服务器端通道的实现类（便于 Netty 做反射处理）
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    // 对服务端的 NioServerSocketChannel 添加 Handler
                    // LoggingHandler 是 netty 内置的一种 ChannelDuplexHandler，
                    // 既可以处理出站事件，又可以处理入站事件，即 LoggingHandler
                    // 既记录出站日志又记录入站日志。
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // 对服务端接收到的、与客户端之间建立的 SocketChannel 添加 Handler
                    .childHandler(new ChannelInitializer() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(serverHandler);
                        }
                    });

            // Start the server.

            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.

            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.

            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

EventLoopGroup的创建与初始化

对应代码

EventLoopGroup bossGroup = new NioEventLoopGroup();

跟踪 NioEventLoopGroup的无参构造

NioEventLoopGroup()
-->
NioEventLoopGroup(int nThreads)
-->
NioEventLoopGroup(int nThreads, Executor executor)
-->
NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider)
-->
NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory)
-->
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
    super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}

这里能看到，如果构造传入的线程数为0，则使用DEFAULT_EVENT_LOOP_THREADS

值为系统变量 io.netty.eventLoopThreads，没有环境变量就取cpu逻辑线程数*2

例如我的电脑为8核16线程，nThreads = 16 * 2

继续跟踪代码，以下代码有部分省略

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                        EventExecutorChooserFactory chooserFactory, Object... args) {
    // 检查线程数量不能小于1
    checkPositive(nThreads, "nThreads");

    // 这里的 ThreadPerTaskExecutor 实例是下文用于创建 EventExecutor 实例的参数
    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }

    children = new EventExecutor[nThreads];

    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
            // 创建EventLoop（重点）
            children[i] = newChild(executor, args);
            success = true;
        } catch (Exception e) {
            // TODO: Think about if this is a good exception type
            throw new IllegalStateException("failed to create a child event loop", e);
        } finally {
            。。。。。。
        }
    }

    // chooser 的作用是为了实现 next()方法，即从 EventLoopGroup 中挑选
    // 一个 NioEventLoop 来处理连接上 IO 事件的方法
    chooser = chooserFactory.newChooser(children);

    。。。。。。
}

ThreadPerTaskExecutor很简单，实现了Executor接口

public final class ThreadPerTaskExecutor implements Executor {
    。。。。。。
    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}

这意味着每次执行executor.execute方法，都会开启一个线程。

EventLoop的创建是在newChild中

// 类NioEventLoopGroup
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    // selector工厂
    SelectorProvider selectorProvider = (SelectorProvider) args[0];
    // 选择策略工厂
    SelectStrategyFactory selectStrategyFactory = (SelectStrategyFactory) args[1];
    // 拒绝执行处理器（任务添加到队列中失败时调用）
    RejectedExecutionHandler rejectedExecutionHandler = (RejectedExecutionHandler) args[2];
    EventLoopTaskQueueFactory taskQueueFactory = null;
    EventLoopTaskQueueFactory tailTaskQueueFactory = null;

    int argsLength = args.length;
    if (argsLength > 3) {
        taskQueueFactory = (EventLoopTaskQueueFactory) args[3];
    }
    if (argsLength > 4) {
        tailTaskQueueFactory = (EventLoopTaskQueueFactory) args[4];
    }
    // 创建NioEventLoop并返回
    return new NioEventLoop(this, executor, selectorProvider,
            selectStrategyFactory.newSelectStrategy(),
            rejectedExecutionHandler, taskQueueFactory, tailTaskQueueFactory);
}

NioEventLoopGroup的创建，初始化了selector工厂，选择策略，拒绝执行处理器等。

并创建了同样线程数的NioEventLoop

服务端引导类 ServerBootstrap的创建与设置

对应代码

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)......

public ServerBootstrap group(EventLoopGroup group) {
    return group(group, group);
}

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
    super.group(parentGroup);
    if (this.childGroup != null) {
        throw new IllegalStateException("childGroup set already");
    }
    this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
    return this;
}

提供了两个设置EventLoopGroup的方法，也就是parentGroup和childGroup可以是同一个group，

而parentGroup对应线程图中的bossGroup，childGroup对应线程图中的workerGroup

public B channel(Class channelClass) {
    return channelFactory(new (
            ObjectUtil.checkNotNull(channelClass, "channelClass")
    ));
}

这里设置的是channel反射工厂，该工厂会使用反射生成 NioServerSocketChannel对象。

创建并绑定channel

对应代码

ChannelFuture f = b.bind(PORT).sync();

准确点说，是负责创建连接（ACCEPT）的channel的创建

AbstractBootstrap#bind(int inetPort) -->
AbstractBootstrap#bind(SocketAddress localAddress) -->
private ChannelFuture doBind(final SocketAddress localAddress) {
    // 初始化 NioServerSocketChannel 的实例，并且将其注册到
    // bossGroup 中的 EvenLoop 中的 Selector 中，initAndRegister()

    // 实例的初始化和注册（此方法是异步的）：
    // (1) 初始化：将handler注册进通道，并执行handler的handlerAdded、channelRegistered方法
    // (2) 将channel注册进selector
    final ChannelFuture regFuture = initAndRegister();
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {
        return regFuture;
    }

    if (regFuture.isDone()) {
        // At this point we know that the registration was complete and successful.

        // 若异步过程 initAndRegister()已经执行完毕，则进入该分支
        ChannelPromise promise = channel.newPromise();
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {
        // Registration future is almost always fulfilled already, but just in case it's not.

        // 若异步过程 initAndRegister()还未执行完毕，则进入该分支
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        regFuture.addListener(new ChannelFutureListener() {
            // 监听 regFuture 的完成事件，完成之后再调用
            // doBind0(regFuture, channel, localAddress, promise);
            @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);
                }
            }
        });
        return promise;
    }
}

initAndRegister方法中主要做了3个操作，channel的创建、初始化以及将channel注册到EventLoop中

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 无参构造会创建pipelile
        // NioServerSocketChannel
        channel = channelFactory.newChannel();
        // 初始化相关属性
        // 如果是ServerBoottrap，还会设置bossGroup的handler，
        // 其中包括ServerBootstrap.handler设置的handler，以及最后添加ServerBootstrapAcceptor
        // ServerBootstrapAcceptor就是将channel注册到workerGroup的类
        init(channel);
    } catch (Throwable t) {
        。。。。。。
    }
    // 将channel注册进selector(监听ACCEPT事件)
    // 依然是通过开启eventLoop线程的方式进行注册
    // MultithreadEventLoopGroup
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}

重新回到AbstractBootstrap#doBind方法中

// 如果上面的initAndRegister方法执行完毕（异步执行的）,则执行doBind0
if (regFuture.isDone()) {
    // At this point we know that the registration was complete and successful.

    // 若异步过程 initAndRegister()已经执行完毕，则进入该分支
    ChannelPromise promise = channel.newPromise();
    doBind0(regFuture, channel, localAddress, promise);
    return promise;
} else {
    // Registration future is almost always fulfilled already, but just in case it's not.

    // 若异步过程 initAndRegister()还未执行完毕，则进入该分支
    final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
    regFuture.addListener(new ChannelFutureListener() {
        // 监听 regFuture 的完成事件，完成之后再调用
        // doBind0(regFuture, channel, localAddress, promise);
        @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);
            }
        }
    });
    return promise;
}

上面这段if/esle做了同一件是，就是自行doBind0方法，区别在于如果initAndRegister执行完毕，则执行调用doBind0，否则添加监听器，等执行完成触发调用doBind0

继续看doBind0

// 类AbstractBootstrap
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.

    // execute方法会将这个Runnable加入到taskQueue中，并开线程执行EventLoop的run方法(死循环)
    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());
            }
        }
    });
}

channel.eventLoop().execute这个后面再说，可以看到，里面的逻辑是调用channel.bind在实现绑定的，继续跟踪

AbstractChannel#bind(SocketAddress localAddress, ChannelPromise promise)
-->
// 类AbstractChannel
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
    // tail就是TailContext
    return tail.bind(localAddress, promise);
}
-->
// 类AbstractChannel
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
    ObjectUtil.checkNotNull(localAddress, "localAddress");
    if (isNotValidPromise(promise, false)) {
        // cancelled
        return promise;
    }
    // 在管道中从当前handlerContext往前查找实现了bind方法的handlerContext
    final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        // 执行handlerContext的bind方法
        next.invokeBind(localAddress, promise);
    } else {
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                next.invokeBind(localAddress, promise);
            }
        }, promise, null, false);
    }
    return promise;
}
-->
// 类AbstractChannel
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
    if (invokeHandler()) {
        try {
            ((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
        } catch (Throwable t) {
            notifyOutboundHandlerException(t, promise);
        }
    } else {
        bind(localAddress, promise);
    }
}

从上面可以看到，最终会执行handler的bind方法，拿 LoggingHandler的bind方法举例

// 类LoggingHandler
public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
    if (logger.isEnabled(internalLevel)) {
        logger.log(internalLevel, format(ctx, "BIND", localAddress));
    }
    ctx.bind(localAddress, promise);
}

ctx.bind(localAddress, promise)是不是很眼熟，没错，就是AbstractChannel#bind(final SocketAddress localAddress, final ChannelPromise promise)

就像一个循环，每一次都在当前handlerContext往前找有实现了bind方法的handlerContext，执行bind，然后继续往前找。

最终找到管道中的第一个handler，也就是 HeadContext，看看它实现的bind方法

// 类HeadContext
public void bind(
        ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
    unsafe.bind(localAddress, promise);
}
-->
AbstractChannel#bind(final SocketAddress localAddress, final ChannelPromise promise)
-->
NioServerSocketChannel#doBind(SocketAddress localAddress)

最后，还是Java NIO的API来绑定

参考资料：

《Netty in Action》，Norman Maurer

《Scalable IO in Java》，Doug Lea

Original: https://www.cnblogs.com/konghuanxi/p/16381317.html
Author: 王谷雨
Title: Netty源码解读（二）-服务端源码讲解