Netty服务端启动源码分析（三十一）

迷南。 2022-10-29 14:23 194阅读 0赞

今天进行 Netty服务端启动源码分析

1、首先服务端代码：

public void start() throws InterruptedException {
        final EchoServerHandler serverHandler = new EchoServerHandler();
        /*线程组*/
        EventLoopGroup group  = new NioEventLoopGroup();
        try {
            /*服务端启动必备*/
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)
                    .channel(NioServerSocketChannel.class)/*指定使用NIO的通信模式*/
            .localAddress(new InetSocketAddress(port))/*指定监听端口*/
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(serverHandler);
                }
            });
            ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞到完成*/
            f.channel().closeFuture().sync();/*阻塞当前线程，直到服务器的ServerChannel被关闭*/
        } finally {
            group.shutdownGracefully().sync();
        }
    }

在服务器启动的常规代码里，首先是实例化 NioEventLoopGroup 和 ServerBootstrap，然后是各种配置。各种配置基本上就是简单的 set。比较特别的是 channel()方法，通过查看代码我们发现， 我们传入的是一个 Class 类型的变量，Netty 将通过反射产生 Channel 的实例。

public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }

继续：

public ReflectiveChannelFactory(Class<? extends T> clazz) {
        if (clazz == null) {
            throw new NullPointerException("clazz");
        }
        this.clazz = clazz;
    }
    
    @Override
    public T newChannel() {
        try {
            return clazz.getConstructor().newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + clazz, t);
        }
    }

2、 **bind()****方法**

所以，关键代码就落在了 bind()方法上。这个方法实际调用的是 AbstractBootstrap 的doBind()方法

/**
         * Create a new {@link Channel} and bind it.
         */
        public ChannelFuture bind() {
            validate();
            SocketAddress localAddress = this.localAddress;
            if (localAddress == null) {
                throw new IllegalStateException("localAddress not set");
            }
            return doBind(localAddress);
        }

doBind()方法中，比较关键的是 initAndRegister 方法和 doBind0 方法，

private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
    
        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            // Registration future is almost always fulfilled already, but just in case it's not.
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    Throwable cause = future.cause();
                    if (cause != null) {
                        // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                        // IllegalStateException once we try to access the EventLoop of the Channel.
                        promise.setFailure(cause);
                    } else {
                        // Registration was successful, so set the correct executor to use.
                        // See https://github.com/netty/netty/issues/2586
                        promise.registered();
    
                        doBind0(regFuture, channel, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }

***3、initAndRegister 方法：***final ChannelFuture regFuture = initAndRegister();

顾名思义，负责初始化和注册，具体做了哪些事呢，其实比较关键的就是三个方法：

final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                // channel can be null if newChannel crashed (eg SocketException("too many open files"))
                channel.unsafe().closeForcibly();
                // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
                return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
            }
            // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
            return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
        }
    
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }
    
        // If we are here and the promise is not failed, it's one of the following cases:
        // 1) If we attempted registration from the event loop, the registration has been completed at this point.
        //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
        // 2) If we attempted registration from the other thread, the registration request has been successfully
        //    added to the event loop's task queue for later execution.
        //    i.e. It's safe to attempt bind() or connect() now:
        //         because bind() or connect() will be executed *after* the scheduled registration task is executed
        //         because register(), bind(), and connect() are all bound to the same thread.
    
        return regFuture;
    }

*4、channel*  =  **channelFactory** .newChannel(); 通过反射 new  出一个主  channel （ NioServerSocketChannel ）的实例，在NioServerSocketChanne 构造方法中表明关注的事件为 OP\_ACCEPT

/**
     * Create a new instance
     */
    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
    
    /**
     * Create a new instance using the given {@link SelectorProvider}.
     */
    public NioServerSocketChannel(SelectorProvider provider) {
        this(newSocket(provider));
    }
    
    /**
     * Create a new instance using the given {@link ServerSocketChannel}.
     */
    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

5、同时拿到一个 JDK 中的 ServerSocketChannel 进行对应

private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            /**
             *  Use the {@link SelectorProvider} to open {@link SocketChannel} and so remove condition in
             *  {@link SelectorProvider#provider()} which is called by each ServerSocketChannel.open() otherwise.
             *
             *  See <a href="https://github.com/netty/netty/issues/2308">#2308</a>.
             */
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException(
                    "Failed to open a server socket.", e);
        }
    }

6、 一路向上，在父类 AbstractNioChannel 里配置关注的事件为 OP_ACCEPT 并且将 NioServerSocketChannel 和 Jdk 里的 ServerSocketChannel 进行挂钩，然后设置当前通道为非阻塞。

public abstract class AbstractNioChannel extends AbstractChannel {
    
        private static final InternalLogger logger =
                InternalLoggerFactory.getInstance(AbstractNioChannel.class);
    
        private static final ClosedChannelException DO_CLOSE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
                new ClosedChannelException(), AbstractNioChannel.class, "doClose()");
    
        private final SelectableChannel ch;
        protected final int readInterestOp;
        volatile SelectionKey selectionKey;
        boolean readPending;
        private final Runnable clearReadPendingRunnable = new Runnable() {
            @Override
            public void run() {
                clearReadPending0();
            }
        };
    
        /**
         * The future of the current connection attempt.  If not null, subsequent
         * connection attempts will fail.
         */
        private ChannelPromise connectPromise;
        private ScheduledFuture<?> connectTimeoutFuture;
        private SocketAddress requestedRemoteAddress;
    
        /**
         * Create a new instance
         *
         * @param parent            the parent {@link Channel} by which this instance was created. May be {@code null}
         * @param ch                the underlying {@link SelectableChannel} on which it operates
         * @param readInterestOp    the ops to set to receive data from the {@link SelectableChannel}
         */
        protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
            super(parent);
            this.ch = ch;
            this.readInterestOp = readInterestOp;
            try {
                ch.configureBlocking(false);
            } catch (IOException e) {
                try {
                    ch.close();
                } catch (IOException e2) {
                    if (logger.isWarnEnabled()) {
                        logger.warn(
                                "Failed to close a partially initialized socket.", e2);
                    }
                }
    
                throw new ChannelException("Failed to enter non-blocking mode.", e);
            }
        }

7、在更顶层的父类 AbstractChannel 的构造方法中会增加主 channel 的 pipeline

protected AbstractChannel(Channel parent, ChannelId id) {
        this.parent = parent;
        this.id = id;
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }

点击：newChannelPipeline方法

protected DefaultChannelPipeline newChannelPipeline() {
            return new DefaultChannelPipeline(this);
        }

8、 这个 pipeline 就是 DefaultChannelPipeline，在 DefaultChannelPipeline 中增加两个缺省的 handler，TailContext 和 HeadContext

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

TailContext 和 HeadContext 从类的声明可以看出，HeadContext 既是 ChannelHandlerContext 又是出站和入站的Handler，

// A special catch-all handler that handles both bytes and messages.
    final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {

9、 TailContext 既是 ChannelHandlerContext 又是入站的 Handler。结合 DefaultChannelPipeline 中的 addLast0 方法

private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;
        newCtx.prev = prev;
        newCtx.next = tail;
        prev.next = newCtx;
        tail.prev = newCtx;
    }

我们可以得出结论，我们增加的任何自定义 handler，一定是放在 TailContext 和 HeadContext 之间，HeadContext 一定是入站的第一个 Handler 和出站的最后一个 Handler，

10、TailContext  是入站的最后一个  Handler 。到这里， newChannel() 执行完成了，一个NioServerSocketChannel 的实例被创建了。 接下来执行 init( *channel* ); init 方法本身是个抽象方法，所以实际调用  ServerBootstrap  的实现

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

源码：

在这个实现中，首先是一系列的局部变量的赋值，然后为这个主  channel ，也就是NioServerSocketChannel 增加  Handler 。如果定义了主 handler  加入主  pipeline ，紧跟着异步加入一个处理客户端新连接的Handler： ServerBootstrapAcceptor

@Override
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }
    
        final Map<AttributeKey<?>, Object> attrs = attrs0();
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                @SuppressWarnings("unchecked")
                AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                channel.attr(key).set(e.getValue());
            }
        }
    
        ChannelPipeline p = channel.pipeline();
    
        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
        }
    
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }
    
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

11、首先看这个  ServerBootstrapAcceptor

private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter

它是  ServerBootstrap  的内部类，并且处理入站事件，主要重写了  channelRead  方法和 exceptionCaught 方法。当这个异步操作执行完，现在主  pipeline  中有三个  handler  了，分别

是  HeadContext->ServerBootstrapAcceptor->TailContext 。 在异步执行机制上，eventLoop.execute  方法实际执行  SingleThreadEventExecutor  的 execute 方法

@Override
    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
    
        boolean inEventLoop = inEventLoop();
        addTask(task);
        if (!inEventLoop) {
            startThread();
            if (isShutdown() && removeTask(task)) {
                reject();
            }
        }
    
        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }

将当前任务加入队列后，会先判定，当前执行 execute 方法的是否为 EventLoop 本身所持有的那个线程，

@Override
    public boolean inEventLoop() {
        return inEventLoop(Thread.currentThread());
    }

继续：

@Override
    public boolean inEventLoop(Thread thread) {
        return thread == this.thread;
    }

如果不是，执行 startThread()方法，其实是执行 doStartThread()方法，最终执行了 executor.execute 方法，从前面的描述中我们知道，每个 NioEventLoop 的执行器为

ThreadPerTaskExecutor ，所以实际执行的是  ThreadPerTaskExecutor  的  execute  方法，

@Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }

在 ThreadPerTaskExecutor 的 execute 方法中启动一个实际的线程，并且在传递给这个execute 方法的 Runnable 实例中，

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 {
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                    logger.warn("Unexpected exception from an event executor: ", t);
                } finally {
                    for (;;) {
                        int oldState = state;
                        if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
                                SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
                            break;
                        }
                    }
    
                    // Check if confirmShutdown() was called at the end of the loop.
                    if (success && gracefulShutdownStartTime == 0) {
                        if (logger.isErrorEnabled()) {
                            logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +
                                    SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must " +
                                    "be called before run() implementation terminates.");
                        }
                    }
    
                    try {
                        // Run all remaining tasks and shutdown hooks.
                        for (;;) {
                            if (confirmShutdown()) {
                                break;
                            }
                        }
                    } finally {
                        try {
                            cleanup();
                        } finally {
                            STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
                            threadLock.release();
                            if (!taskQueue.isEmpty()) {
                                if (logger.isWarnEnabled()) {
                                    logger.warn("An event executor terminated with " +
                                            "non-empty task queue (" + taskQueue.size() + ')');
                                }
                            }
    
                            terminationFuture.setSuccess(null);
                        }
                    }
                }
            }
        });
    }

12、将 SingleThreadEventExecutor 持有的 Thread 会和刚刚 new 出来的线程（也就是当前执行 **executor**.execute 方法的实际线程）进行挂钩，于是每个 NioEventLoop 有了自己的独有线 程。并且执行的 run 方法，从具体的实现类来说其实就是 NioEventLoop，而 NioEventLoop 的 run 方法中，所做的事情就是不断的进行 select 和事件集的处理以及其他系统任务的处理。 init(*channel*);执行完成，接着执行 config().group().register(*channel*);方法，但这是个抽象 方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

执行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

具体执行谁？从类的继承关系来说，应该是  MultithreadEventLoopGroup 。调试一下，可以发现，确实进入了 MultithreadEventLoopGroup  的  register(Channel channel) 方法。

![20210215165213256.png][]

点击

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

next() 会在多个  Loop  中挑选一个  EventLoop ， register(channel) 则会调用进入SingleThreadEventLoop 的  register() 方法，并最终进入  AbstractChannel  的内部抽象类

AbstractUnsafe  的  register  方法，首先是一系列的检查

@Override
    public ChannelFuture register(final ChannelPromise promise) {
        ObjectUtil.checkNotNull(promise, "promise");
        promise.channel().unsafe().register(this, promise);
        return promise;
    }

进入

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

源码：

@Override
    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        if (eventLoop == null) {
            throw new NullPointerException("eventLoop");
        }
        if (isRegistered()) {
            promise.setFailure(new IllegalStateException("registered to an event loop already"));
            return;
        }
        if (!isCompatible(eventLoop)) {
            promise.setFailure(
                    new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
            return;
        }
    
        AbstractChannel.this.eventLoop = eventLoop;
    
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            try {
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        register0(promise);
                    }
                });
            } catch (Throwable t) {
                logger.warn(
                        "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                        AbstractChannel.this, t);
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }
    }

还是会先判定，当前执行 register 方法的是否为 EventLoop 本身所持有的那个线程，是则直接执行 register0 方法，否则依然在一个 Runnable 中执行 register0 方法。 register0 方法实际调用了在 AbstractNioChannel 中的 doRegister()注册了一个键值为 0 的 事件，那么也就意味着，其实到了这个时候，还没真正开始监听网络连接事件。 但是到了这一步，initAndRegister()方法的主要工作已经做完了，接下来就是 doBind0 方法。

***13、doBind0***  ***方法***

这个方法，做了实际绑定

private static void doBind0(
            final ChannelFuture regFuture, final Channel channel,
            final SocketAddress localAddress, final ChannelPromise promise) {
    
        // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
        // the pipeline in its channelRegistered() implementation.
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                if (regFuture.isSuccess()) {
                    channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                } else {
                    promise.setFailure(regFuture.cause());
                }
            }
        });
    }

点击bind：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel 类中：

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

调用了  DefaultChannelPipeline  的  bind  方法， bind  是个出站事件，所以由出站  handler 进行处理，tail  开始往出站方向进行传播，目前来说，只有  head  会处理出站事件，所以由 head 来处理，而最终进入了  AbstractChannel  进行绑定。

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

继续：

@Override
    public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        if (isNotValidPromise(promise, false)) {
            // cancelled
            return promise;
        }
    
        final AbstractChannelHandlerContext next = findContextOutbound();
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeBind(localAddress, promise);
        } else {
            safeExecute(executor, new Runnable() {
                @Override
                public void run() {
                    next.invokeBind(localAddress, promise);
                }
            }, promise, null);
        }
        return promise;
    }

findContextOutbound 方法用来寻找出站 Handler，如何寻找的？

private AbstractChannelHandlerContext findContextOutbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    }

以当前 handler 的 ChannelHandlerContext 为起点，往自己的前驱 Context 方向寻找，一 直找到一个出站 Handler 为止。因为这里只有 HeadContext 会处理出站事件，所以由

HeadContext  来处理。最终进入了 AbstractChanne. AbstractUnsafe  进行绑定

@Override
    public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
        assertEventLoop();
    
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
    
        // See: https://github.com/netty/netty/issues/576
        if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
            localAddress instanceof InetSocketAddress &&
            !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
            !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
            // Warn a user about the fact that a non-root user can't receive a
            // broadcast packet on *nix if the socket is bound on non-wildcard address.
            logger.warn(
                    "A non-root user can't receive a broadcast packet if the socket " +
                    "is not bound to a wildcard address; binding to a non-wildcard " +
                    "address (" + localAddress + ") anyway as requested.");
        }
    
        boolean wasActive = isActive();
        try {
            doBind(localAddress);
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            closeIfClosed();
            return;
        }
    //绑定完成后激活 ChannelActive 事件，
        if (!wasActive && isActive()) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    pipeline.fireChannelActive();
                }
            });
        }
    
        safeSetSuccess(promise);
    }

14、调用由 NioServerSocketChanne 实现的 doBind 方法，当然实际是调用  jdk  的方法进行了绑定

@Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

由 pipeline.fireChannelActive();这个方法通知 pipeline，channel 被激活，事件开始在pipeline 上流动，ChannelActive 是个入站事件，会按照 head->tail 的顺序执行 Inbound 处理

器，  ServerBootstrapAcceptor  和  tail  的  channelActive  方法都没有做任何实质性的事情，只有head 有实质性代码

@Override
    public final ChannelPipeline fireChannelActive() {
        AbstractChannelHandlerContext.invokeChannelActive(head);
        return this;
    }

继续：

static void invokeChannelActive(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelActive();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelActive();
                }
            });
        }
    }

继续：

private void invokeChannelActive() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelActive(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelActive();
        }
    }

继续：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6][]

源码：

@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    
        readIfIsAutoRead();
    }

继续：

private void readIfIsAutoRead() {
        if (channel.config().isAutoRead()) {
            channel.read();
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7][]

源码：

@Override
        public Channel read() {
            pipeline.read();
            return this;
        }

调用了  AbstractChannel  的  read  方法，其中又调用了  DefaultChannelPipeline  的  read  方法，其实就是调用了 tail  的  read() 方法

@Override
    public final ChannelPipeline read() {
        tail.read();
        return this;
    }

继续：abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext, ResourceLeakHint

@Override
    public ChannelHandlerContext read() {
        final AbstractChannelHandlerContext next = findContextOutbound();
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeRead();
        } else {
            Runnable task = next.invokeReadTask;
            if (task == null) {
                next.invokeReadTask = task = new Runnable() {
                    @Override
                    public void run() {
                        next.invokeRead();
                    }
                };
            }
            executor.execute(task);
        }
    
        return this;
    }

15、 Read 是一个 Outbound 事件，因此 findContextOutbound()会按照 tail->.....->head 寻找前面最近的一个 Outbound 处理器，目前有三个处理器：tail->ServerBootstrapAcceptor->head， 但只有 head 是 outbound 处理器，所以，head 的 read 方法中

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8][]

源码：

@Override
    public void read(ChannelHandlerContext ctx) {
        unsafe.beginRead();
    }

进入了 AbstractChanne. AbstractUnsafe 的 beginRead 方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 9][]

源码：

@Override
    public final void beginRead() {
        assertEventLoop();
    
        if (!isActive()) {
            return;
        }
    
        try {
            doBeginRead();
        } catch (final Exception e) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    pipeline.fireExceptionCaught(e);
                }
            });
            close(voidPromise());
        }
    }

而其中的最终 doBeginRead()调用的就是 AbstractChannel 的子类 AbstractNioChannel 的方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 10][]

源码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 11][]

源码：

@Override
    protected void doBeginRead() throws Exception {
        // Channel.read() or ChannelHandlerContext.read() was called
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return;
        }
    
        readPending = true;
    
        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

会将 NioServerSocketChanne 感兴趣的事件设置为 OP\_ACCEPT(16)。到了这一步 AbstractBootstrap 的 doBind 方法执行完成，服务器的启动也算完成了。

到此，Netty服务端启动源码分析完成，下篇分析Netty各种事件的原理，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/a8890ba74a314457ac93d4e04e3d8a3b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/abde37f16936441db30e279324781895.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/d70fbb3d32394d6287b517fd327a787c.png
[20210215165213256.png]: /images/20221024/1a09d06a7cd64de0bd7d090d6301f45d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/7f6bc5450e894500bf0f7e7ff543b416.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/480fa2d2e8e74e0cbca8309af5ca5c49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: /images/20221024/24061e9328654c5db4dc64c13ac84700.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6]: /images/20221024/bf5438387980444fb81048fec6133279.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7]: /images/20221024/9ea25b98cc074a629ffdfa6cbc743eaa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8]: /images/20221024/8ae7b8ed95714cca814d5baf0a9401fa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 9]: /images/20221024/d162d5b0fbaa4c4c8f5b3eee0556b5a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 10]: /images/20221024/10ad7183a91b4f3db2d8231eddc4340d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 11]: /images/20221024/d40d7ba5a9aa40759eeec42df28a92d6.png