详细图解 Netty Reactor 启动全流程 | 万字长文 | 多图预警

本系列Netty源码解析文章基于 4.1.56.Final版本

大家第一眼看到这幅流程图，是不是脑瓜子嗡嗡的呢？

大家先不要惊慌，问题不大，本文笔者的目的就是要让大家清晰的理解这幅流程图，从而深刻的理解Netty Reactor的启动全流程，包括其中涉及到的各种代码设计实现细节。

在上篇文章《聊聊Netty那些事儿之Reactor在Netty中的实现(创建篇)》中我们详细介绍了Netty服务端核心引擎组件 主从Reactor组模型 NioEventLoopGroup以及 Reactor模型 NioEventLoop的创建过程。最终我们得到了netty Reactor模型的运行骨架如下：

现在Netty服务端程序的骨架是搭建好了，本文我们就基于这个骨架来深入剖析下Netty服务端的启动过程。

我们继续回到上篇文章提到的Netty服务端代码模板中，在创建完主从Reactor线程组: bossGroup， workerGroup后，接下来就开始配置Netty服务端的启动辅助类 ServerBootstrap了。

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

    public static void main(String[] args) throws Exception {
        // Configure the server.

        //创建主从Reactor线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)//配置主从Reactor
             .channel(NioServerSocketChannel.class)//配置主Reactor中的channel类型
             .option(ChannelOption.SO_BACKLOG, 100)//设置主Reactor中channel的option选项
             .handler(new LoggingHandler(LogLevel.INFO))//设置主Reactor中Channel->pipline->handler
             .childHandler(new ChannelInitializer() {//设置从Reactor中注册channel的pipeline
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

            // Start the server. 绑定端口启动服务，开始监听accept事件
            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();
        }
    }
}

在上篇文章中我们对代码模板中涉及到 ServerBootstrap的一些配置方法做了简单的介绍，大家如果忘记的话，可以在返回去回顾一下。

ServerBootstrap类其实没有什么特别的逻辑，主要是对Netty启动过程中需要用到的一些核心信息进行配置管理，比如：

• Netty的核心引擎组件 主从Reactor线程组： bossGroup，workerGroup。通过 ServerBootstrap#group方法配置。
• Netty服务端使用到的Channel类型： NioServerSocketChannel ,通过 ServerBootstrap#channel方法配置。
  以及配置 NioServerSocketChannel时用到的 SocketOption。 SocketOption用于设置底层JDK NIO Socket的一些选项。通过 ServerBootstrap#option方法进行配置。

主ReactorGroup中的MainReactor管理的Channel类型为 NioServerSocketChannel，如图所示主要用来监听端口，接收客户端连接，为客户端创建初始化 NioSocketChannel，然后采用 round-robin轮询的方式从图中从ReactorGroup中选择一个SubReactor与该客户端 NioSocketChannel进行绑定。

从ReactorGroup中的SubReactor管理的Channel类型为 NioSocketChannel，它是netty中定义客户端连接的一个模型，每个连接对应一个。如图所示SubReactor负责监听处理绑定在其上的所有 NioSocketChannel上的IO事件。

• 保存服务端 NioServerSocketChannel和客户端 NioSocketChannel对应 pipeline中指定的 ChannelHandler。用于后续Channel向Reactor注册成功之后，初始化Channel里的pipeline。

不管是服务端用到的 NioServerSocketChannel还是客户端用到的 NioSocketChannel，每个 Channel实例都会有一个 Pipeline， Pipeline中有多个 ChannelHandler用于编排处理对应 Channel上感兴趣的 IO事件。

ServerBootstrap结构中包含了netty服务端程序启动的所有配置信息，在我们介绍启动流程之前，先来看下 ServerBootstrap的源码结构：

ServerBootstrap

ServerBootstrap的继承结构比较简单，继承层次的职责分工也比较明确。

ServerBootstrap主要负责对 主从Reactor线程组相关的配置进行管理，其中带 child前缀的配置方法是对 从Reactor线程组的相关配置管理。 从Reactor线程组中的 Sub Reactor负责管理的客户端 NioSocketChannel相关配置存储在 ServerBootstrap结构中。

父类 AbstractBootstrap则是主要负责对 主Reactor线程组相关的配置进行管理，以及 主Reactor线程组中的 Main Reactor负责处理的服务端 ServerSocketChannel相关的配置管理。

1. 配置主从Reactor线程组

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)//配置主从Reactor

public class ServerBootstrap extends AbstractBootstrap {

     //Main Reactor线程组
    volatile EventLoopGroup group;
    //Sub Reactor线程组
    private volatile EventLoopGroup childGroup;

    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        //父类管理主Reactor线程组
        super.group(parentGroup);
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        }
        this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
        return this;
    }

}

2. 配置服务端ServerSocketChannel

ServerBootstrap b = new ServerBootstrap();
b.channel(NioServerSocketChannel.class);

public class ServerBootstrap extends AbstractBootstrap {

    //用于创建ServerSocketChannel  ReflectiveChannelFactory
    private volatile ChannelFactory channelFactory;

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

    @Deprecated
    public B channelFactory(ChannelFactory channelFactory) {
        ObjectUtil.checkNotNull(channelFactory, "channelFactory");
        if (this.channelFactory != null) {
            throw new IllegalStateException("channelFactory set already");
        }

        this.channelFactory = channelFactory;
        return self();
    }

}

在向 ServerBootstrap配置服务端 ServerSocketChannel的 channel方法中，其实是创建了一个 ChannelFactory工厂实例 ReflectiveChannelFactory,在Netty服务端启动的过程中，会通过这个 ChannelFactory去创建相应的 Channel实例。

我们可以通过这个方法来配置netty的IO模型，下面为 ServerSocketChannel在不同IO模型下的实现：

BIO NIO AIO OioServerSocketChannel NioServerSocketChannel AioServerSocketChannel

EventLoopGroupReactor线程组在不同IO模型下的实现：

BIO NIO AIO ThreadPerChannelEventLoopGroup NioEventLoopGroup AioEventLoopGroup

我们只需要将 IO模型的这些核心接口对应的实现类 前缀改为对应 IO模型的前缀，就可以轻松在Netty中完成对 IO模型的切换。

2.1 ReflectiveChannelFactory

public class ReflectiveChannelFactory implements ChannelFactory {
    //NioServerSocketChannelde 构造器
    private final Constructor constructor;

    public ReflectiveChannelFactory(Class clazz) {
        ObjectUtil.checkNotNull(clazz, "clazz");
        try {
            //反射获取NioServerSocketChannel的构造器
            this.constructor = clazz.getConstructor();
        } catch (NoSuchMethodException e) {
            throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                    " does not have a public non-arg constructor", e);
        }
    }

    @Override
    public T newChannel() {
        try {
            //创建NioServerSocketChannel实例
            return constructor.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
        }
    }
}

从类的签名我们可以看出，这个工厂类是通过 泛型加 反射的方式来创建对应的 Channel实例。

• 泛型参数 T extends Channel表示的是要通过工厂类创建的 Channel类型，这里我们初始化的是 NioServerSocketChannel。
• 在 ReflectiveChannelFactory的构造器中通过 反射的方式获取 NioServerSocketChannel的构造器。
• 在 newChannel方法中通过构造器反射创建 NioServerSocketChannel实例。

注意这时只是配置阶段， NioServerSocketChannel此时并未被创建。它是在启动的时候才会被创建出来。

3. 为NioServerSocketChannel配置ChannelOption

ServerBootstrap b = new ServerBootstrap();
//设置被MainReactor管理的NioServerSocketChannel的Socket选项
b.option(ChannelOption.SO_BACKLOG, 100)

public abstract class AbstractBootstrap, C extends Channel> implements Cloneable {

    //serverSocketChannel中的ChannelOption配置
    private final Map, Object> options = new LinkedHashMap, Object>();

    public  B option(ChannelOption option, T value) {
        ObjectUtil.checkNotNull(option, "option");
        synchronized (options) {
            if (value == null) {
                options.remove(option);
            } else {
                options.put(option, value);
            }
        }
        return self();
    }
}

无论是服务端的 NioServerSocketChannel还是客户端的 NioSocketChannel它们的相关底层Socket选项 ChannelOption配置全部存放于一个 Map类型的数据结构中。

由于客户端 NioSocketChannel是由 从Reactor线程组中的 Sub Reactor来负责处理，所以涉及到客户端 NioSocketChannel所有的方法和配置全部是以 child前缀开头。

ServerBootstrap b = new ServerBootstrap();
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)

public class ServerBootstrap extends AbstractBootstrap {

   //客户端SocketChannel对应的ChannelOption配置
    private final Map, Object> childOptions = new LinkedHashMap, Object>();

    public  ServerBootstrap childOption(ChannelOption childOption, T value) {
        ObjectUtil.checkNotNull(childOption, "childOption");
        synchronized (childOptions) {
            if (value == null) {
                childOptions.remove(childOption);
            } else {
                childOptions.put(childOption, value);
            }
        }
        return this;
    }
}

相关的底层Socket选项，netty全部枚举在ChannelOption类中，笔者这里就不一一列举了，在本系列后续相关的文章中，笔者还会为大家详细的介绍这些参数的作用。

public class ChannelOption extends AbstractConstant> {

    ..................省略..............

    public static final ChannelOption SO_BROADCAST = valueOf("SO_BROADCAST");
    public static final ChannelOption SO_KEEPALIVE = valueOf("SO_KEEPALIVE");
    public static final ChannelOption SO_SNDBUF = valueOf("SO_SNDBUF");
    public static final ChannelOption SO_RCVBUF = valueOf("SO_RCVBUF");
    public static final ChannelOption SO_REUSEADDR = valueOf("SO_REUSEADDR");
    public static final ChannelOption SO_LINGER = valueOf("SO_LINGER");
    public static final ChannelOption SO_BACKLOG = valueOf("SO_BACKLOG");
    public static final ChannelOption SO_TIMEOUT = valueOf("SO_TIMEOUT");

    ..................省略..............

}

4. 为服务端NioServerSocketChannel中的Pipeline配置ChannelHandler

    //serverSocketChannel中pipeline里的handler(主要是acceptor)
    private volatile ChannelHandler handler;

    public B handler(ChannelHandler handler) {
        this.handler = ObjectUtil.checkNotNull(handler, "handler");
        return self();
    }

向 NioServerSocketChannel中的 Pipeline添加 ChannelHandler分为两种方式：

• 显式添加： 显式添加的方式是由用户在main线程中通过 ServerBootstrap#handler的方式添加。如果需要添加多个 ChannelHandler，则可以通过 ChannelInitializer向 pipeline中进行添加。

关于 ChannelInitializer后面笔者会有详细介绍，这里大家只需要知道 ChannelInitializer是一种特殊的 ChannelHandler，用于初始化 pipeline。适用于向pipeline中添加多个ChannelHandler的场景。

            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)//配置主从Reactor
             .channel(NioServerSocketChannel.class)//配置主Reactor中的channel类型
             .handler(new ChannelInitializer() {
                 @Override
                 protected void initChannel(NioServerSocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(channelhandler1)
                      .addLast(channelHandler2)

                      ......

                      .addLast(channelHandler3);
                 }
             })

• 隐式添加：隐式添加主要添加的就是 主ReactorGroup的核心组件也就是下图中的 acceptor，Netty中的实现为 ServerBootstrapAcceptor，本质上也是一种 ChannelHandler，主要负责在客户端连接建立好后，初始化客户端 NioSocketChannel，在 从Reactor线程组中选取一个 Sub Reactor，将客户端 NioSocketChannel 注册到 Sub Reactor中的 selector上。

隐式添加 ServerBootstrapAcceptor是由Netty框架在启动的时候负责添加，用户无需关心。

在本例中， NioServerSocketChannel的 PipeLine中只有两个 ChannelHandler,一个由用户在外部显式添加的 LoggingHandler,另一个是由Netty框架隐式添加的 ServerBootstrapAcceptor。

其实我们在实际项目使用的过程中，不会向netty服务端 NioServerSocketChannel添加额外的ChannelHandler， NioServerSocketChannel只需要专心做好自己最重要的本职工作接收客户端连接就好了。这里额外添加一个 LoggingHandler只是为了向大家展示 ServerBootstrap的配置方法。

5. 为客户端NioSocketChannel中的Pipeline配置ChannelHandler

            final EchoServerHandler serverHandler = new EchoServerHandler();

            serverBootstrap.childHandler(new ChannelInitializer() {//设置从Reactor中注册channel的pipeline
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();

                     p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

    //socketChannel中pipeline中的处理handler
    private volatile ChannelHandler childHandler;

    public ServerBootstrap childHandler(ChannelHandler childHandler) {
        this.childHandler = ObjectUtil.checkNotNull(childHandler, "childHandler");
        return this;
    }

向客户端 NioSocketChannel中的 Pipeline里添加 ChannelHandler完全是由用户自己控制显式添加，添加的数量不受限制。

由于在Netty的 IO线程模型中，是由单个 Sub Reactor线程负责执行客户端 NioSocketChannel中的 Pipeline，一个 Sub Reactor线程负责处理多个 NioSocketChannel上的 IO事件，如果 Pipeline中的 ChannelHandler添加的太多，就会影响 Sub Reactor线程执行其他 NioSocketChannel上的 Pipeline，从而降低 IO处理效率，降低吞吐量。

所以 Pipeline中的 ChannelHandler不易添加过多，并且不能再 ChannelHandler中执行耗时的业务处理任务。

在我们通过 ServerBootstrap配置netty服务端启动信息的时候，无论是向服务端 NioServerSocketChannel的pipeline中添加ChannelHandler，还是向客户端 NioSocketChannel的pipeline中添加ChannelHandler，当涉及到多个ChannelHandler添加的时候，我们都会用到 ChannelInitializer，那么这个 ChannelInitializer究竟是何方圣神，为什么要这样做呢？我们接着往下看~~

ChannelInitializer

首先 ChannelInitializer它继承于 ChannelHandler，它自己本身就是一个ChannelHandler，所以它可以添加到 childHandler中。

其他的父类大家这里可以不用管，后面文章中笔者会一一为大家详细介绍。

那为什么不直接添加 ChannelHandler 而是选择用 ChannelInitializer 呢？

这里主要有两点原因：

• 前边我们提到，客户端 NioSocketChannel是在服务端accept连接后，在服务端 NioServerSocketChannel中被创建出来的。 但是此时我们正处于配置 ServerBootStrap 阶段，服务端还没有启动，更没有客户端连接上来，此时客户端 NioSocketChannel还没有被创建出来，所以也就没办法向客户端 NioSocketChannel的pipeline中添加 ChannelHandler。
• 客户端 NioSocketChannel中 Pipeline里可以添加任意多个 ChannelHandler，但是Netty框架无法预知用户到底需要添加多少个 ChannelHandler，所以Netty框架提供了回调函数 ChannelInitializer#initChannel，使用户可以自定义 ChannelHandler的添加行为。

当客户端 NioSocketChannel注册到对应的 Sub Reactor上后，紧接着就会初始化 NioSocketChannel中的 Pipeline，此时Netty框架会回调 ChannelInitializer#initChannel执行用户自定义的添加逻辑。

public abstract class ChannelInitializer extends ChannelInboundHandlerAdapter {

    @Override
    @SuppressWarnings("unchecked")
    public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        //当channelRegister事件发生时，调用initChannel初始化pipeline
        if (initChannel(ctx)) {
                 .................省略...............

        } else {
                 .................省略...............

        }
    }

    private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
        if (initMap.add(ctx)) { // Guard against re-entrance.

            try {
                //此时客户单NioSocketChannel已经创建并初始化好了
                initChannel((C) ctx.channel());
            } catch (Throwable cause) {
                 .................省略...............

            } finally {
                  .................省略...............

            }
            return true;
        }
        return false;
    }

    protected abstract void initChannel(C ch) throws Exception;

    .................省略...............

}

这里由netty框架回调的 ChannelInitializer#initChannel方法正是我们自定义的添加逻辑。

            final EchoServerHandler serverHandler = new EchoServerHandler();

            serverBootstrap.childHandler(new ChannelInitializer() {//设置从Reactor中注册channel的pipeline
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();

                     p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

到此为止，Netty服务端启动所需要的必要配置信息，已经全部存入 ServerBootStrap启动辅助类中。

接下来要做的事情就是服务端的启动了。

// Start the server. 绑定端口启动服务，开始监听accept事件
ChannelFuture f = serverBootStrap.bind(PORT).sync();

Netty服务端的启动

经过前面的铺垫终于来到了本文的核心内容—-Netty服务端的启动过程。

如代码模板中的示例所示，Netty服务端的启动过程封装在 io.netty.bootstrap.AbstractBootstrap#bind(int)函数中。

接下来我们看一下Netty服务端在启动过程中究竟干了哪些事情？

大家看到这副启动流程图先不要慌，接下来的内容笔者会带大家各个击破它，在文章的最后保证让大家看懂这副流程图。

我们先来从netty服务端启动的入口函数开始我们今天的源码解析旅程：

    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }

    public ChannelFuture bind(SocketAddress localAddress) {
        //校验Netty核心组件是否配置齐全
        validate();
        //服务端开始启动，绑定端口地址，接收客户端连接
        return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
    }

   private ChannelFuture doBind(final SocketAddress localAddress) {
        //异步创建，初始化，注册ServerSocketChannel到main reactor上
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }

        if (regFuture.isDone()) {

           ........serverSocketChannel向Main Reactor注册成功后开始绑定端口....,

        } else {
            //如果此时注册操作没有完成，则向regFuture添加operationComplete回调函数，注册成功后回调。
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {

                   ........serverSocketChannel向Main Reactor注册成功后开始绑定端口....,
            });
            return promise;
        }
    }

Netty服务端的启动流程总体如下：

• 创建服务端 NioServerSocketChannel并初始化。
• 将服务端 NioServerSocketChannel注册到 主Reactor线程组中。
• 注册成功后，开始初始化 NioServerSocketChannel中的pipeline，然后在pipeline中触发channelRegister事件。
• 随后由 NioServerSocketChannel绑定端口地址。
• 绑定端口地址成功后，向 NioServerSocketChannel对应的 Pipeline中触发传播 ChannelActive事件，在 ChannelActive事件回调中向 Main Reactor注册 OP_ACCEPT事件，开始等待客户端连接。服务端启动完成。

当netty服务端启动成功之后，最终我们会得到如下结构的阵型，开始枕戈待旦，准备接收客户端的连接，Reactor开始运转。

接下来，我们就来看下Netty源码是如何实现以上步骤的~~

1. initAndRegister

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            //创建NioServerSocketChannel
            //ReflectiveChannelFactory通过泛型，反射，工厂的方式灵活创建不同类型的channel
            channel = channelFactory.newChannel();
            //初始化NioServerSocketChannel
            init(channel);
        } catch (Throwable t) {
            ..............省略.................

        }

        //向MainReactor注册ServerSocketChannel
        ChannelFuture regFuture = config().group().register(channel);

           ..............省略.................

        return regFuture;
    }

从函数命名中我们可以看出，这个函数主要做的事情就是首先创建 NioServerSocketChannel，并对 NioServerSocketChannel进行初始化，最后将 NioServerSocketChannel注册到 Main Reactor中。

1.1 创建NioServerSocketChannel

还记得我们在介绍 ServerBootstrap启动辅助类配置服务端 ServerSocketChannel类型的时候提到的工厂类 ReflectiveChannelFactory吗？

因为当时我们在配置 ServerBootstrap启动辅助类的时候，还没到启动阶段，而配置阶段并不是创建具体 ServerSocketChannel的时机。

所以Netty通过 工厂模式将要创建的 ServerSocketChannel的类型（通过泛型指定）以及 创建的过程(封装在newChannel函数中)统统先封装在工厂类 ReflectiveChannelFactory中。

ReflectiveChannelFactory通过 泛型， 反射， 工厂的方式 灵活创建不同类型的 channel

等待创建时机来临，我们调用保存在 ServerBootstrap中的 channelFactory直接进行创建。

public class ReflectiveChannelFactory implements ChannelFactory {

    private final Constructor constructor;

    @Override
    public T newChannel() {
        try {
            return constructor.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
        }
    }
}

下面我们来看下 NioServerSocketChannel的构建过程：

1.1.1 NioServerSocketChannel

public class NioServerSocketChannel extends AbstractNioMessageChannel
                             implements io.netty.channel.socket.ServerSocketChannel {

    //SelectorProvider(用于创建Selector和Selectable Channels)
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();

    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }

    //创建JDK NIO ServerSocketChannel
    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException(
                    "Failed to open a server socket.", e);
        }
    }

     //ServerSocketChannel相关的配置
    private final ServerSocketChannelConfig config;

    public NioServerSocketChannel(ServerSocketChannel channel) {
        //父类AbstractNioChannel中保存JDK NIO原生ServerSocketChannel以及要监听的事件OP_ACCEPT
        super(null, channel, SelectionKey.OP_ACCEPT);
        //DefaultChannelConfig中设置用于Channel接收数据用的buffer->AdaptiveRecvByteBufAllocator
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

}

• 首先调用 newSocket创建JDK NIO 原生 ServerSocketChannel，这里调用了 SelectorProvider#openServerSocketChannel来创建JDK NIO 原生 ServerSocketChannel，我们在上篇文章《聊聊Netty那些事儿之Reactor在Netty中的实现(创建篇)》中详细的介绍了 SelectorProvider相关内容，当时是用 SelectorProvider来创建 Reactor中的 Selector。大家还记得吗？？
• 通过父类构造器设置 NioServerSocketChannel感兴趣的 IO事件,这里设置的是 SelectionKey.OP_ACCEPT事件。并将JDK NIO 原生 ServerSocketChannel封装起来。
• 创建 Channel的配置类 NioServerSocketChannelConfig，在配置类中封装了对 Channel底层的一些配置行为，以及JDK中的 ServerSocket。以及创建 NioServerSocketChannel接收数据用的 Buffer分配器 AdaptiveRecvByteBufAllocator。

NioServerSocketChannelConfig没什么重要的东西，我们这里也不必深究，它就是管理 NioServerSocketChannel相关的配置，这里唯一需要大家注意的是这个用于 Channel接收数据用的 Buffer分配器AdaptiveRecvByteBufAllocator，我们后面在介绍Netty如何接收连接的时候还会提到。

NioServerSocketChannel的整体构建过程介绍完了，现在我们来按照继承层次再回过头来看下 NioServerSocketChannel的层次构建，来看下每一层都创建了什么，封装了什么，这些信息都是 Channel的核心信息，所以有必要了解一下。

在 NioServerSocketChannel的创建过程中，我们主要关注继承结构图中红框标注的三个类，其他的我们占时先不用管。

其中 AbstractNioMessageChannel类主要是对 NioServerSocketChannel底层读写行为的封装和定义，比如accept接收客户端连接。这个我们后续会介绍到，这里我们并不展开。

1.1.2 AbstractNioChannel

public abstract class AbstractNioChannel extends AbstractChannel {
   //JDK NIO原生Selectable Channel
    private final SelectableChannel ch;
    // Channel监听事件集合 这里是SelectionKey.OP_ACCEPT事件
    protected final int readInterestOp;

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            //设置Channel为非阻塞 配合IO多路复用模型
            ch.configureBlocking(false);
        } catch (IOException e) {
            .............省略................

        }
    }
}

• 封装由 SelectorProvider创建出来的JDK NIO原生 ServerSocketChannel。
• 封装 Channel在创建时指定感兴趣的 IO事件，对于 NioServerSocketChannel来说感兴趣的 IO事件为 OP_ACCEPT事件。
• 设置JDK NIO原生 ServerSocketChannel为非阻塞模式， 配合IO多路复用模型。

1.1.3 AbstractChannel

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {

    //channel是由创建层次的，比如ServerSocketChannel 是 SocketChannel的 parent
    private final Channel parent;
    //channel全局唯一ID machineId+processId+sequence+timestamp+random
    private final ChannelId id;
    //unsafe用于封装对底层socket的相关操作
    private final Unsafe unsafe;
    //为channel分配独立的pipeline用于IO事件编排
    private final DefaultChannelPipeline pipeline;

    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        //channel全局唯一ID machineId+processId+sequence+timestamp+random
        id = newId();
        //unsafe用于定义实现对Channel的底层操作
        unsafe = newUnsafe();
        //为channel分配独立的pipeline用于IO事件编排
        pipeline = newChannelPipeline();
    }
}

• Netty中的 Channel创建是有层次的，这里的 parent属性用来保存上一级的 Channel，比如这里的 NioServerSocketChannel是顶级 Channel，所以它的 parent = null。客户端 NioSocketChannel是由 NioServerSocketChannel创建的，所以它的 parent = NioServerSocketChannel。
• 为 Channel分配全局唯一的 ChannelId。 ChannelId由机器Id(machineId)，进程Id（ processId），序列号（ sequence），时间戳（ timestamp），随机数（ random）构成

   private DefaultChannelId() {
        data = new byte[MACHINE_ID.length + PROCESS_ID_LEN + SEQUENCE_LEN + TIMESTAMP_LEN + RANDOM_LEN];
        int i = 0;

        // machineId
        System.arraycopy(MACHINE_ID, 0, data, i, MACHINE_ID.length);
        i += MACHINE_ID.length;

        // processId
        i = writeInt(i, PROCESS_ID);

        // sequence
        i = writeInt(i, nextSequence.getAndIncrement());

        // timestamp (kind of)
        i = writeLong(i, Long.reverse(System.nanoTime()) ^ System.currentTimeMillis());

        // random
        int random = PlatformDependent.threadLocalRandom().nextInt();
        i = writeInt(i, random);
        assert i == data.length;

        hashCode = Arrays.hashCode(data);
    }

• 创建 NioServerSocketChannel的底层操作类 Unsafe。这里创建的是 io.netty.channel.nio.AbstractNioMessageChannel.NioMessageUnsafe。

Unsafe为 Channel接口的一个内部接口，用于定义实现对Channel底层的各种操作， Unsafe接口定义的操作行为只能由Netty框架的 Reactor线程调用，用户线程禁止调用。

interface Unsafe {

        //分配接收数据用的Buffer
        RecvByteBufAllocator.Handle recvBufAllocHandle();

        //服务端绑定的端口地址
        SocketAddress localAddress();
        //远端地址
        SocketAddress remoteAddress();
        //channel向Reactor注册
        void register(EventLoop eventLoop, ChannelPromise promise);

        //服务端绑定端口地址
        void bind(SocketAddress localAddress, ChannelPromise promise);
        //客户端连接服务端
        void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
        //关闭channle
        void close(ChannelPromise promise);
        //读数据
        void beginRead();
        //写数据
        void write(Object msg, ChannelPromise promise);

    }

• 为 NioServerSocketChannel分配独立的 pipeline用于IO事件编排。 pipeline其实是一个 ChannelHandlerContext类型的双向链表。头结点 HeadContext,尾结点 TailContext。 ChannelHandlerContext中包装着 ChannelHandler。

ChannelHandlerContext保存 ChannelHandler上下文信息，用于事件传播。后面笔者会单独开一篇文章介绍，这里我们还是聚焦于启动主线。

这里只是为了让大家简单理解 pipeline的一个大致的结构，后面会写一篇文章专门详细讲解 pipeline。

    protected DefaultChannelPipeline(Channel channel) {
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);

        tail = new TailContext(this);
        head = new HeadContext(this);

        head.next = tail;
        tail.prev = head;
    }

到了这里 NioServerSocketChannel就创建完毕了，我们来回顾下它到底包含了哪些核心信息。

1.2 初始化NioServerSocketChannel

   void init(Channel channel) {
        //向NioServerSocketChannelConfig设置ServerSocketChannelOption
        setChannelOptions(channel, newOptionsArray(), logger);
        //向netty自定义的NioServerSocketChannel设置attributes
        setAttributes(channel, attrs0().entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY));

        ChannelPipeline p = channel.pipeline();

        //获取从Reactor线程组
        final EventLoopGroup currentChildGroup = childGroup;
        //获取用于初始化客户端NioSocketChannel的ChannelInitializer
        final ChannelHandler currentChildHandler = childHandler;
        //获取用户配置的客户端SocketChannel的channelOption以及attributes
        final Entry, Object>[] currentChildOptions;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(EMPTY_OPTION_ARRAY);
        }
        final Entry, Object>[] currentChildAttrs = childAttrs.entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY);

        //向NioServerSocketChannel中的pipeline添加初始化ChannelHandler的逻辑
        p.addLast(new ChannelInitializer() {
            @Override
            public void initChannel(final Channel ch) {
                final ChannelPipeline pipeline = ch.pipeline();
                //ServerBootstrap中用户指定的channelHandler
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    //LoggingHandler
                    pipeline.addLast(handler);
                }
                //添加用于接收客户端连接的acceptor
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

• 向 NioServerSocketChannelConfig设置 ServerSocketChannelOption。
• 向netty自定义的 NioServerSocketChannel设置 ChannelAttributes

Netty自定义的 SocketChannel类型均继承 AttributeMap接口以及 DefaultAttributeMap类，正是它们定义了 ChannelAttributes。用于向 Channel添加用户自定义的一些信息。

这个 ChannelAttributes的用处大有可为，Netty后边的许多特性都是依靠这个 ChannelAttributes来实现的。这里先卖个关子，大家可以自己先想一下可以用这个 ChannelAttributes做哪些事情？

• 获取从Reactor线程组 childGroup，以及用于初始化客户端 NioSocketChannel的 ChannelInitializer, ChannelOption, ChannelAttributes，这些信息均是由用户在启动的时候向 ServerBootstrap添加的客户端 NioServerChannel配置信息。这里用这些信息来初始化 ServerBootstrapAcceptor。因为后续会在 ServerBootstrapAcceptor中接收客户端连接以及创建 NioServerChannel。
• 向 NioServerSocketChannel中的 pipeline添加用于初始化 pipeline的 ChannelInitializer。

问题来了，这里为什么不干脆直接将 ChannelHandler 添加到 pipeline 中，而是又使用到了 ChannelInitializer 呢？

其实原因有两点：

• 为了保证 线程安全地初始化 pipeline，所以初始化的动作需要由 Reactor线程进行，而当前线程是 用户程序的 启动Main线程 并 不是Reactor线程。这里不能立即初始化。
• 初始化 Channel中 pipeline的动作，需要等到 Channel注册到对应的 Reactor中才可以进行初始化，当前只是创建好了 NioServerSocketChannel，但并未注册到 Main Reactor上。

初始化 NioServerSocketChannel中 pipeline的时机是：当 NioServerSocketChannel注册到 Main Reactor之后，绑定端口地址之前。

前边在介绍 ServerBootstrap配置 childHandler时也用到了 ChannelInitializer，还记得吗？？

问题又来了，大家注意下 ChannelInitializer#initChannel 方法，在该初始化回调方法中，添加LoggingHandler是直接向pipeline中添加，而添加Acceptor为什么不是直接添加而是封装成异步任务呢？

这里先给大家卖个关子，笔者会在后续流程中为大家解答~~~~~

此时 NioServerSocketChannel中的 pipeline结构如下图所示：

1.3 向Main Reactor注册NioServerSocketChannel

从 ServerBootstrap获取主Reactor线程组 NioEventLoopGroup，将 NioServerSocketChannel注册到 NioEventLoopGroup中。

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

下面我们来看下具体的注册过程：

1.3.1 主Reactor线程组中选取一个Main Reactor进行注册

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

    @Override
    public EventExecutor next() {
        return chooser.next();
    }

    //获取绑定策略
    @Override
    public EventExecutorChooser newChooser(EventExecutor[] executors) {
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }

    //采用轮询round-robin的方式选择Reactor
    @Override
    public EventExecutor next() {
            return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];
    }

Netty通过 next()方法根据上篇文章《聊聊Netty那些事儿之Reactor在Netty中的实现(创建篇)》提到的 channel到reactor的绑定策略，从 ReactorGroup中选取一个Reactor进行注册绑定。之后 Channel生命周期内的所有 IO 事件都由这个 Reactor 负责处理，如 accept、connect、read、write等 IO 事件。

一个 channel只能绑定到一个 Reactor上，一个 Reactor负责监听 多个channel。

由于这里是 NioServerSocketChannle向 Main Reactor进行注册绑定，所以 Main Reactor主要负责处理的 IO事件是 OP_ACCEPT事件。

1.3.2 向绑定后的Main Reactor进行注册

向 Reactor进行注册的行为定义在 NioEventLoop的父类 SingleThreadEventLoop中，印象模糊的同学可以在回看下上篇文章中的 NioEventLoop继承结构小节内容。

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {

    @Override
    public ChannelFuture register(Channel channel) {
        //注册channel到绑定的Reactor上
        return register(new DefaultChannelPromise(channel, this));
    }

    @Override
    public ChannelFuture register(final ChannelPromise promise) {
        ObjectUtil.checkNotNull(promise, "promise");
        //unsafe负责channel底层的各种操作
        promise.channel().unsafe().register(this, promise);
        return promise;
    }
}

通过 NioServerSocketChannel中的 Unsafe类执行底层具体的注册动作。

protected abstract class AbstractUnsafe implements Unsafe {

        /**
         * 注册Channel到绑定的Reactor上
         * */
        @Override
        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            ObjectUtil.checkNotNull(eventLoop, "eventLoop");
            if (isRegistered()) {
                promise.setFailure(new IllegalStateException("registered to an event loop already"));
                return;
            }
            //EventLoop的类型要与Channel的类型一样  Nio Oio Aio
            if (!isCompatible(eventLoop)) {
                promise.setFailure(
                        new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
                return;
            }

            //在channel上设置绑定的Reactor
            AbstractChannel.this.eventLoop = eventLoop;

            /**
             * 执行channel注册的操作必须是Reactor线程来完成
             *
             * 1: 如果当前执行线程是Reactor线程，则直接执行register0进行注册
             * 2：如果当前执行线程是外部线程，则需要将register0注册操作 封装程异步Task 由Reactor线程执行
             * */
            if (eventLoop.inEventLoop()) {
                register0(promise);
            } else {
                try {
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            register0(promise);
                        }
                    });
                } catch (Throwable t) {
                   ...............省略...............

                }
            }
        }
}

• 首先检查 NioServerSocketChannel是否已经完成注册。如果以完成注册，则直接设置代表注册操作结果的 ChannelPromise为 fail状态。
• 通过 isCompatible方法验证Reactor模型 EventLoop是否与 Channel的类型匹配。 NioEventLoop对应于 NioServerSocketChannel。

上篇文章我们介绍过 Netty对三种 IO模型： Oio,Nio,Aio的支持，用户可以通过改变Netty核心类的前缀轻松切换 IO模型。 isCompatible方法目的就是需要保证 Reactor和 Channel使用的是同一种 IO模型。

• 在 Channel中保存其绑定的 Reactor实例。
• 执行 Channel向 Reactor注册的动作必须要确保是在 Reactor线程中执行。
• 如果当前线程是 Reactor线程则直接执行注册动作 register0
• 如果当前线程不是 Reactor线程，则需要将注册动作 register0封装成异步任务，存放在 Reactor中的 taskQueue中，等待 Reactor线程执行。

当前执行线程并不是 Reactor线程，而是用户程序的启动线程 Main线程。

1.3.3 Reactor线程的启动

上篇文章中我们在介绍 NioEventLoopGroup的创建过程中提到了一个构造器参数 executor，它用于启动 Reactor线程，类型为 ThreadPerTaskExecutor。

当时笔者向大家卖了一个关子~~ “Reactor线程是何时启动的?”

那么现在就到了为大家揭晓谜底的时候了~~

Reactor线程 的启动是在向 Reactor 提交第一个异步任务的时候启动的。

Netty中的主Reactor线程组 NioEventLoopGroup中的Main Reactor NioEventLoop是在用户程序 Main线程向 Main Reactor提交用于注册 NioServerSocketChannel的异步任务时开始启动。

   eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            register0(promise);
                        }
                    });

接下来我们关注下 NioEventLoop的 execute方法

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {

    @Override
    public void execute(Runnable task) {
        ObjectUtil.checkNotNull(task, "task");
        execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));
    }

    private void execute(Runnable task, boolean immediate) {
        //当前线程是否为Reactor线程
        boolean inEventLoop = inEventLoop();
        //addTaskWakesUp = true  addTask唤醒Reactor线程执行任务
        addTask(task);
        if (!inEventLoop) {
            //如果当前线程不是Reactor线程，则启动Reactor线程
            //这里可以看出Reactor线程的启动是通过 向NioEventLoop添加异步任务时启动的
            startThread();

            .....................省略.....................

        }
        .....................省略.....................

    }

}

• 首先将异步任务 task添加到 Reactor中的 taskQueue中。
• 判断当前线程是否为 Reactor线程，此时当前执行线程为用户程序启动线程，所以这里调用 startThread启动 Reactor线程。

1.3.4 startThread

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    //定义Reactor线程状态
    private static final int ST_NOT_STARTED = 1;
    private static final int ST_STARTED = 2;
    private static final int ST_SHUTTING_DOWN = 3;
    private static final int ST_SHUTDOWN = 4;
    private static final int ST_TERMINATED = 5;

     //Reactor线程状态  初始为 未启动状态
    private volatile int state = ST_NOT_STARTED;

    //Reactor线程状态字段state 原子更新器
    private static final AtomicIntegerFieldUpdater STATE_UPDATER =
    AtomicIntegerFieldUpdater.newUpdater(SingleThreadEventExecutor.class, "state");

    private void startThread() {
        if (state == ST_NOT_STARTED) {
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                boolean success = false;
                try {
                    doStartThread();
                    success = true;
                } finally {
                    if (!success) {
                        STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
                    }
                }
            }
        }
    }

}

• Reactor线程初始化状态为 ST_NOT_STARTED,首先 CAS更新状态为 ST_STARTED
• doStartThread启动 Reactor线程
• 启动失败的话，需要将 Reactor线程状态改回 ST_NOT_STARTED

    //ThreadPerTaskExecutor 用于启动Reactor线程
    private final Executor executor;

    private void doStartThread() {
        assert thread == null;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }

                boolean success = false;
                updateLastExecutionTime();
                try {
                    //Reactor线程开始启动
                    SingleThreadEventExecutor.this.run();
                    success = true;
                }

                ................省略..............

        }

这里就来到了 ThreadPerTaskExecutor类型的 executor的用武之地了。

• Reactor线程的核心工作之前介绍过： 轮询所有注册其上的Channel中的IO就绪事件， 处理对应Channel上的IO事件， 执行异步任务。Netty将这些核心工作封装在 io.netty.channel.nio.NioEventLoop#run方法中。

• 将 NioEventLoop#run封装在异步任务中，提交给 executor执行， Reactor线程至此开始工作了就。

public final class ThreadPerTaskExecutor implements Executor {
    private final ThreadFactory threadFactory;

    @Override
    public void execute(Runnable command) {
        //启动Reactor线程
        threadFactory.newThread(command).start();
    }
}

此时 Reactor线程已经启动， 后面的工作全部都由这个 Reactor线程 来负责执行了。

而用户启动线程在向 Reactor提交完 NioServerSocketChannel的注册任务 register0后，就逐步退出调用堆栈，回退到最开始的启动入口处 ChannelFuture f = b.bind(PORT).sync()。

此时 Reactor中的任务队列中只有一个任务 register0， Reactor线程启动后，会从任务队列中取出任务执行。

至此 NioServerSocketChannel的注册工作正式拉开帷幕~~

1.3.5 register0

       //true if the channel has never been registered, false otherwise
        private boolean neverRegistered = true;

        private void register0(ChannelPromise promise) {
            try {
                //查看注册操作是否已经取消，或者对应channel已经关闭
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                boolean firstRegistration = neverRegistered;
                //执行真正的注册操作
                doRegister();
                //修改注册状态
                neverRegistered = false;
                registered = true;
                //回调pipeline中添加的ChannelInitializer的handlerAdded方法，在这里初始化channelPipeline
                pipeline.invokeHandlerAddedIfNeeded();
                //设置regFuture为success，触发operationComplete回调,将bind操作放入Reactor的任务队列中，等待Reactor线程执行。
                safeSetSuccess(promise);
                //触发channelRegister事件
                pipeline.fireChannelRegistered();
                //对于服务端ServerSocketChannel来说 只有绑定端口地址成功后 channel的状态才是active的。
                //此时绑定操作作为异步任务在Reactor的任务队列中，绑定操作还没开始，所以这里的isActive()是false
                if (isActive()) {
                    if (firstRegistration) {
                        //触发channelActive事件
                        pipeline.fireChannelActive();
                    } else if (config().isAutoRead()) {
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                 ............省略.............

            }
        }

register0是驱动整个 Channel注册绑定流程的关键方法，下面我们来看下它的核心逻辑：

• 首先需要检查 Channel的注册动作是否在 Reactor线程外被取消了已经 !promise.setUncancellable()。检查要注册的 Channel是否已经关闭 !ensureOpen(promise)。如果 Channel已经关闭或者注册操作已经被取消，那么就直接返回，停止注册流程。
• 调用 doRegister()方法，执行真正的注册操作。最终实现在 AbstractChannel的子类 AbstractNioChannel中，这个我们一会在介绍，先关注整体流程。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {

   /**
     * Is called after the {@link Channel} is registered with its {@link EventLoop} as part of the register process.

     *
     * Sub-classes may override this method
     */
    protected void doRegister() throws Exception {
        // NOOP
    }

}

• 当 Channel向 Reactor注册完毕后，调用 pipeline.invokeHandlerAddedIfNeeded()方法，触发回调pipeline中添加的ChannelInitializer的handlerAdded方法，在handlerAdded方法中利用前面提到的 ChannelInitializer初始化 ChannelPipeline。

初始化 ChannelPipeline的时机是当 Channel向对应的 Reactor注册成功后，在 handlerAdded事件回调中利用 ChannelInitializer进行初始化。

• 设置 regFuture为 Success，并回调注册在 regFuture上的 ChannelFutureListener#operationComplete方法，在 operationComplete回调方法中将 绑定操作封装成异步任务，提交到 Reactor的 taskQueue中。等待 Reactor的执行。

还记得这个 regFuture在哪里出现的吗？它是在哪里被创建，又是在哪里添加的 ChannelFutureListener呢？ 大家还有印象吗？回忆不起来也没关系，笔者后面还会提到

• 通过 pipeline.fireChannelRegistered()在 pipeline中触发 channelRegister事件。

pipeline中 channelHandler的 channelRegistered方法被回调。

• 对于Netty服务端 NioServerSocketChannel来说， 只有 绑定端口地址成功后 channel的状态才是 active的。此时 绑定操作在 regFuture上注册的 ChannelFutureListener#operationComplete回调方法中被作为异步任务提交到了 Reactor的任务队列中， Reactor线程还 没开始执行 绑定任务。所以这里的 isActive()是 false。

当 Reactor线程执行完 register0方法后，才会去执行 绑定任务。

下面我们来看下 register0方法中这些 核心步骤的具体实现：

1.3.6 doRegister()

public abstract class AbstractNioChannel extends AbstractChannel {

    //channel注册到Selector后获得的SelectKey
    volatile SelectionKey selectionKey;

    @Override
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) {
                ...............省略....................

            }
        }
    }

}

调用底层 JDK NIO Channel方法 java.nio.channels.SelectableChannel#register(java.nio.channels.Selector, int, java.lang.Object)，将Netty NioServerSocketChannel中包装的 JDK NIO ServerSocketChannel注册到 Reactor中的 JDK NIO Selector上。

简单介绍下 SelectableChannel#register方法参数的含义：

• Selector：表示 JDK NIO Channel将要向哪个 Selector进行注册。
• int ops： 表示 Channel上感兴趣的 IO事件，当对应的 IO事件就绪时， Selector会返回 Channel对应的 SelectionKey。

SelectionKey可以理解为 Channel在 Selector上的特殊表示形式， SelectionKey中封装了 Channel感兴趣的 IO事件集合~~~interestOps，以及 IO就绪的事件集合~~readyOps， 同时也封装了对应的 JDK NIO Channel以及注册的 Selector。最后还有一个重要的属性 attachment，可以允许我们在 SelectionKey上附加一些自定义的对象。

• Object attachment：向 SelectionKey中添加用户自定义的附加对象。

这里 NioServerSocketChannel向 Reactor中的 Selector注册的 IO事件为 0，这个操作的主要目的是先获取到 Channel在 Selector中对应的 SelectionKey，完成注册。当绑定操作完成后，在去向 SelectionKey添加感兴趣的 IO事件~~~ OP_ACCEPT事件。

同时通过 SelectableChannel#register方法将Netty自定义的 NioServerSocketChannel（这里的 this指针）附着在 SelectionKey的 attechment属性上， 完成Netty自定义 Channel 与JDK NIO Channel 的关系绑定。这样在每次对 Selector进行 IO就绪事件轮询时，Netty 都可以从 JDK NIO Selector返回的 SelectionKey中获取到自定义的 Channel对象（这里指的就是 NioServerSocketChannel）。

1.3.7 HandlerAdded事件回调中初始化ChannelPipeline

当 NioServerSocketChannel注册到 Main Reactor上的 Selector后，Netty通过调用 pipeline.invokeHandlerAddedIfNeeded()开始回调 NioServerSocketChannel中 pipeline里的ChannelHandler的 handlerAdded方法。

此时 NioServerSocketChannel的 pipeline结构如下：

此时 pipeline中只有在初始化 NioServerSocketChannel时添加的 ChannelInitializer。

我们来看下 ChannelInitializer中 handlerAdded回调方法具体作了哪些事情~~

public abstract class ChannelInitializer extends ChannelInboundHandlerAdapter {

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isRegistered()) {
            if (initChannel(ctx)) {
                //初始化工作完成后，需要将自身从pipeline中移除
                removeState(ctx);
            }
        }
    }

    //ChannelInitializer实例是被所有的Channel共享的，用于初始化ChannelPipeline
    //通过Set集合保存已经初始化的ChannelPipeline，避免重复初始化同一ChannelPipeline
    private final Set initMap = Collections.newSetFromMap(
            new ConcurrentHashMap());

    private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
        if (initMap.add(ctx)) { // Guard against re-entrance.

            try {
                initChannel((C) ctx.channel());
            } catch (Throwable cause) {
                exceptionCaught(ctx, cause);
            } finally {
                ChannelPipeline pipeline = ctx.pipeline();
                if (pipeline.context(this) != null) {
                     //初始化完毕后，从pipeline中移除自身
                    pipeline.remove(this);
                }
            }
            return true;
        }
        return false;
    }

    //匿名类实现，这里指定具体的初始化逻辑
    protected abstract void initChannel(C ch) throws Exception;

    private void removeState(final ChannelHandlerContext ctx) {
        //从initMap防重Set集合中删除ChannelInitializer
        if (ctx.isRemoved()) {
            initMap.remove(ctx);
        } else {
            ctx.executor().execute(new Runnable() {
                @Override
                public void run() {
                    initMap.remove(ctx);
                }
            });
        }
    }
}

ChannelInitializer中的初始化逻辑比较简单明了：

• 首先要判断必须是当前 Channel已经完成注册后，才可以进行 pipeline的初始化。 ctx.channel().isRegistered()
• 调用 ChannelInitializer的匿名类指定的 initChannel执行自定义的初始化逻辑。

        p.addLast(new ChannelInitializer() {
            @Override
            public void initChannel(final Channel ch) {
                final ChannelPipeline pipeline = ch.pipeline();
                //ServerBootstrap中用户指定的channelHandler
                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));
                    }
                });
            }
        });

还记得在初始化 NioServerSocketChannel时。 io.netty.bootstrap.ServerBootstrap#init方法中向 pipeline中添加的 ChannelInitializer吗？

• 当执行完 initChannel 方法后， ChannelPipeline的初始化就结束了，此时 ChannelInitializer就没必要再继续呆在 pipeline中了，所需要将 ChannelInitializer从 pipeline中删除。 pipeline.remove(this)

当初始化完 pipeline时，此时 pipeline的结构再次发生了变化：

此时 Main Reactor中的任务队列 taskQueue结构变化为：

添加 ServerBootstrapAcceptor的任务是在初始化 NioServerSocketChannel的时候向main reactor提交过去的。还记得吗？

1.3.8 回调regFuture的ChannelFutureListener

在本小节《Netty服务端的启动》的最开始，我们介绍了服务端启动的入口函数 io.netty.bootstrap.AbstractBootstrap#doBind，在函数的最开头调用了 initAndRegister()方法用来创建并初始化 NioServerSocketChannel，之后便会将 NioServerSocketChannel注册到 Main Reactor中。

注册的操作是一个异步的过程，所以在 initAndRegister()方法调用后返回一个代表注册结果的 ChannelFuture regFuture。

public abstract class AbstractBootstrap, C extends Channel> implements Cloneable {

    private ChannelFuture doBind(final SocketAddress localAddress) {
        //异步创建，初始化，注册ServerSocketChannel
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }

        if (regFuture.isDone()) {
            //如果注册完成，则进行绑定操作
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            //添加注册完成 回调函数
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {

                         ...............省略...............

                          // 注册完成后，Reactor线程回调这里
                        doBind0(regFuture, channel, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }
}

之后会向 ChannelFuture regFuture添加一个 注册完成后的回调函数~~~~ ChannelFutureListener。在回调函数 operationComplete中开始发起 绑端口地址流程。

那么这个回调函数在什么时候？什么地方发起的呢？？

让我们在回到本小节的主题 register0方法的流程中：

当调用 doRegister()方法完成 NioServerSocketChannel向 Main Reactor的注册后，紧接着会调用 pipeline.invokeHandlerAddedIfNeeded()方法中触发 ChannelInitializer#handlerAdded回调中对 pipeline进行初始化。

最后在 safeSetSuccess方法中，开始回调注册在 regFuture上的 ChannelFutureListener。

   protected final void safeSetSuccess(ChannelPromise promise) {
        if (!(promise instanceof VoidChannelPromise) && !promise.trySuccess()) {
           logger.warn("Failed to mark a promise as success because it is done already: {}", promise);
        }
   }

   @Override
    public boolean trySuccess() {
        return trySuccess(null);
    }

    @Override
    public boolean trySuccess(V result) {
        return setSuccess0(result);
    }

   private boolean setSuccess0(V result) {
        return setValue0(result == null ? SUCCESS : result);
    }

    private boolean setValue0(Object objResult) {
        if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
            RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
            if (checkNotifyWaiters()) {
                //回调注册在promise上的listeners
                notifyListeners();
            }
            return true;
        }
        return false;
    }

safeSetSuccess的逻辑比较简单，首先设置 regFuture结果为 success，并且回调注册在 regFuture上的 ChannelFutureListener。

需要提醒的是，执行 safeSetSuccess方法，以及后边回调 regFuture上的 ChannelFutureListener 这些动作都是由 Reactor线程 执行的。

关于Netty中的 Promise模型后边我会在写一篇专门的文章进行分析，这里大家只需清楚大体的流程即可。不必在意过多的细节。

下面我们把视角切换到 regFuture上的 ChannelFutureListener回调中，看看在 Channel注册完成后，Netty又会做哪些事情？

2. doBind0

public abstract class AbstractBootstrap, C extends Channel> implements Cloneable {

    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());
                }
            }
        });
    }

}

这里Netty又将 绑定端口地址的操作封装成异步任务，提交给 Reactor执行。

但是这里有一个问题，其实此时执行 doBind0 方法的线程正是 Reactor线程 ，那为什么不直接在这里去执行 bind操作 ，而是再次封装成异步任务提交给 Reactor 中的 taskQueue 呢？

反正最终都是由 Reactor线程 执行，这其中又有什么分别呢？

经过上小节的介绍我们知道， bind0方法的调用是由 io.netty.channel.AbstractChannel.AbstractUnsafe#register0方法在将 NioServerSocketChannel注册到 Main Reactor之后，并且 NioServerSocketChannel的 pipeline已经初始化完毕后，通过 safeSetSuccess方法回调过来的。

这个过程全程是由 Reactor线程来负责执行的，但是此时 register0方法并没有执行完毕，还需要执行后面的逻辑。

而绑定逻辑需要在注册逻辑执行完之后执行，所以在 doBind0方法中 Reactor线程会将 绑定操作封装成异步任务先提交给 taskQueue中保存，这样可以使 Reactor线程立马从 safeSetSuccess中返回，继续执行剩下的 register0方法逻辑。

        private void register0(ChannelPromise promise) {
            try {
                ................省略............

                doRegister();
                pipeline.invokeHandlerAddedIfNeeded();
                safeSetSuccess(promise);
                //触发channelRegister事件
                pipeline.fireChannelRegistered();

                if (isActive()) {
                     ................省略............

                }
            } catch (Throwable t) {
                  ................省略............

            }
        }

当 Reactor线程执行完 register0方法后，就会从 taskQueue中取出异步任务执行。

此时 Reactor线程中的 taskQueue结构如下：

• Reactor线程会先取出位于 taskQueue队首的任务执行，这里是指向 NioServerSocketChannel的 pipeline中添加 ServerBootstrapAcceptor的异步任务。

此时 NioServerSocketChannel中 pipeline的结构如下：

• Reactor线程执行绑定任务。

3. 绑定端口地址

对 Channel的操作行为全部定义在 ChannelOutboundInvoker接口中。

public interface ChannelOutboundInvoker {

    /**
     * Request to bind to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation
     * completes, either because the operation was successful or because of an error.

     *
     */
    ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);
}

bind方法由子类 AbstractChannel实现。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {

   @Override
    public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
        return pipeline.bind(localAddress, promise);
    }

}

调用 pipeline.bind(localAddress, promise)在 pipeline中传播 bind事件，触发回调 pipeline中所有 ChannelHandler的 bind方法。

事件在 pipeline中的传播具有方向性：

• inbound事件从 HeadContext开始逐个向后传播直到 TailContext。
• outbound事件则是反向传播，从 TailContext开始反向向前传播直到 HeadContext。

inbound事件只能被 pipeline中的 ChannelInboundHandler响应处理
outbound事件只能被 pipeline中的 ChannelOutboundHandler响应处理

然而这里的 bind事件在Netty中被定义为 outbound事件，所以它在 pipeline中是反向传播。先从 TailContext开始反向传播直到 HeadContext。

然而 bind的核心逻辑也正是实现在 HeadContext中。

3.1 HeadContext

  final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler {

     @Override
        public void bind(
                ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
            //触发AbstractChannel->bind方法 执行JDK NIO SelectableChannel 执行底层绑定操作
            unsafe.bind(localAddress, promise);
        }

}

在 HeadContext#bind回调方法中，调用 Channel里的 unsafe操作类执行真正的绑定操作。

protected abstract class AbstractUnsafe implements Unsafe {

      @Override
        public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
            .................省略................

            //这时channel还未激活  wasActive = false
            boolean wasActive = isActive();
            try {
                //io.netty.channel.socket.nio.NioServerSocketChannel.doBind
                //调用具体channel实现类
                doBind(localAddress);
            } catch (Throwable t) {
                .................省略................

                return;
            }

            //绑定成功后 channel激活 触发channelActive事件传播
            if (!wasActive && isActive()) {
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        //pipeline中触发channelActive事件
                        pipeline.fireChannelActive();
                    }
                });
            }
            //回调注册在promise上的ChannelFutureListener
            safeSetSuccess(promise);
        }

        protected abstract void doBind(SocketAddress localAddress) throws Exception;
}

• 首先执行子类 NioServerSocketChannel具体实现的 doBind方法，通过 JDK NIO 原生 ServerSocketChannel执行底层的绑定操作。

    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        //调用JDK NIO 底层SelectableChannel 执行绑定操作
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

• 判断是否为首次绑定，如果是的话将 触发pipeline中的ChannelActive事件封装成异步任务放入 Reactor中的 taskQueue中。
• 执行 safeSetSuccess(promise),回调注册在 promise上的 ChannelFutureListener。

还是同样的问题，当前执行线程已经是 Reactor线程 了，那么为何不直接触发 pipeline 中的 ChannelActive 事件而是又封装成异步任务呢？？

因为如果直接在这里触发 ChannelActive事件，那么 Reactor线程就会去执行 pipeline中的 ChannelHandler的 channelActive事件回调。

这样的话就影响了 safeSetSuccess(promise)的执行， 延迟了注册在 promise上的 ChannelFutureListener的回调。

到现在为止，Netty服务端就已经完成了绑定端口地址的操作， NioServerSocketChannel的状态现在变为 Active。

最后还有一件重要的事情要做，我们接着来看 pipeline中对 channelActive事件处理。

3.2 channelActive事件处理

channelActive事件在Netty中定义为 inbound事件，所以它在 pipeline中的传播为正向传播，从 HeadContext一直到 TailContext为止。

在 channelActive事件回调中需要触发向 Selector指定需要监听的 IO事件~~ OP_ACCEPT事件。

这块的逻辑主要在 HeadContext中实现。

    final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler {

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            //pipeline中继续向后传播channelActive事件
            ctx.fireChannelActive();
            //如果是autoRead 则自动触发read事件传播
            //在read回调函数中 触发OP_ACCEPT注册
            readIfIsAutoRead();
        }

        private void readIfIsAutoRead() {
            if (channel.config().isAutoRead()) {
                //如果是autoRead 则触发read事件传播
                channel.read();
            }
        }

        //AbstractChannel
        public Channel read() {
                //触发read事件
                pipeline.read();
                return this;
        }

       @Override
        public void read(ChannelHandlerContext ctx) {
            //触发注册OP_ACCEPT或者OP_READ事件
            unsafe.beginRead();
        }
   }

• 在 HeadContext中的 channelActive回调中触发 pipeline中的 read事件。
• 当 read事件再次传播到 HeadContext时，触发 HeadContext#read方法的回调。在 read回调中调用 channel底层操作类 unsafe的 beginRead方法向 selector注册监听 OP_ACCEPT事件。

3.3 beginRead

protected abstract class AbstractUnsafe implements Unsafe {

     @Override
        public final void beginRead() {
            assertEventLoop();
            //channel必须是Active
            if (!isActive()) {
                return;
            }

            try {
                // 触发在selector上注册channel感兴趣的监听事件
                doBeginRead();
            } catch (final Exception e) {
               .............省略..............

            }
        }
}

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    //子类负责继承实现
    protected abstract void doBeginRead() throws Exception;

}

• 断言判断执行该方法的线程必须是 Reactor线程。
• 此时 NioServerSocketChannel已经完成端口地址的绑定操作， isActive() = true
• 调用 doBeginRead实现向 Selector注册监听事件 OP_ACCEPT

public abstract class AbstractNioChannel extends AbstractChannel {

    //channel注册到Selector后获得的SelectKey
    volatile SelectionKey selectionKey;
    // Channel监听事件集合
    protected final int readInterestOp;

    @Override
    protected void doBeginRead() throws Exception {

        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return;
        }

        readPending = true;

        final int interestOps = selectionKey.interestOps();
        /**
         * 1：ServerSocketChannel 初始化时 readInterestOp设置的是OP_ACCEPT事件
         * */
        if ((interestOps & readInterestOp) == 0) {
            //添加OP_ACCEPT事件到interestOps集合中
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }
}

• 前边提到在 NioServerSocketChannel在向 Main Reactor中的 Selector注册后，会获得一个 SelectionKey。这里首先要获取这个 SelectionKey。
• 从 SelectionKey中获取 NioServerSocketChannel感兴趣的 IO事件集合 interestOps，当时在注册的时候 interestOps设置为 0。
• 将在 NioServerSocketChannel初始化时设置的 readInterestOp = OP_ACCEPT，设置到 SelectionKey中的 interestOps集合中。这样 Reactor中的 Selector就开始监听 interestOps集合中包含的 IO事件了。

Main Reactor中主要监听的是 OP_ACCEPT事件。

流程走到这里，Netty服务端就真正的启动起来了，下一步就开始等待接收客户端连接了。大家此刻在来回看这副启动流程图，是不是清晰了很多呢？

此时Netty的 Reactor模型结构如下：

总结

本文我们通过图解源码的方式完整地介绍了整个Netty服务端启动流程，并介绍了在启动过程中涉及到的 ServerBootstrap相关的属性以及配置方式。 NioServerSocketChannel的创建初始化过程以及类的继承结构。

其中重点介绍了 NioServerSocketChannel向 Reactor的注册过程以及 Reactor线程的启动时机和 pipeline的初始化时机。

最后介绍了 NioServerSocketChannel绑定端口地址的整个流程。

上述介绍的这些流程全部是异步操作，各种回调绕来绕去的，需要反复回想下，读异步代码就是这样，需要理清各种回调之间的关系， 并且时刻提醒自己当前的执行线程是什么？

好了，现在Netty服务端已经启动起来，接着就该接收客户端连接了，我们下篇文章见~~~~

Original: https://www.cnblogs.com/binlovetech/p/16442598.html
Author: bin的技术小屋
Title: 详细图解 Netty Reactor 启动全流程 | 万字长文 | 多图预警