//serverSocketChannel中pipeline里的handler(主要是acceptor)    private volatile ChannelHandler handler;    public B handler(ChannelHandler handler) {        this.handler = ObjectUtil.checkNotNull(handler, "handler");        return self();    }

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

关于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);                 }             })

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

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

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

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它继承于ChannelHandler，它自己本身就是一个ChannelHandler，所以它可以添加到childHandler中。

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

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

这里主要有两点原因：

当客户端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服务端的启动过程封装在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服务端的启动流程总体如下：

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

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

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中。

还记得我们在介绍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的构建过程：

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

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

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

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

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

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) {            .............省略................        }    }}

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

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

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

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就创建完毕了，我们来回顾下它到底包含了哪些核心信息。

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

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

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

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

其实原因有两点：

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

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

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

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

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

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

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

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

@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事件。

向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) {                   ...............省略...............                }            }        }}

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

在Channel中保存其绑定的Reactor实例。

执行Channel向Reactor注册的动作必须要确保是在Reactor线程中执行。

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

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

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

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

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

Netty中的主Reactor线程组NioEventLoopGroup中的Main ReactorNioEventLoop是在用户程序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();            .....................省略.....................        }        .....................省略.....................    }}

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

//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的用武之地了。

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的注册工作正式拉开帷幕。

//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注册绑定流程的关键方法，下面我们来看下它的核心逻辑：

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    }}

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

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

pipeline中channelHandler的channelRegistered方法被回调。

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

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

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)，将NettyNioServerSocketChannel中包装的JDK NIO ServerSocketChannel注册到Reactor中的JDK NIO Selector上。

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

SelectionKey可以理解为Channel在Selector上的特殊表示形式， SelectionKey中封装了Channel感兴趣的IO事件集合~interestOps，以及IO就绪的事件集合~~readyOps， 同时也封装了对应的JDK NIO Channel以及注册的Selector。最后还有一个重要的属性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)。

当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中的初始化逻辑比较简单明了：

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吗?

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

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

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

在本小节《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又会做哪些事情?

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

private void register0(ChannelPromise promise) {            try {                ................省略............                doRegister();                pipeline.invokeHandlerAddedIfNeeded();                safeSetSuccess(promise);                //触发channelRegister事件                pipeline.fireChannelRegistered();                if (isActive()) {                     ................省略............                }            } catch (Throwable t) {                  ................省略............            }        }

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

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {   @Override    public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {        return pipeline.bind(localAddress, promise);    }}

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

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

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

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

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

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