netty源码分析（二）

一、ServerBootstrap的初始化

    ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.WARN))
                .childHandler(new MyServerInitializer());  

1、group方法-将创建好的event group赋给ServerBootstrap的属性

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        //调用父类的group方法
        super.group(parentGroup);
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        }
        //workgroup 赋值给了 childGroup
        this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
        return this;
    }
    
    public B group(EventLoopGroup group) {
        ObjectUtil.checkNotNull(group, "group");
        if (this.group != null) {
            throw new IllegalStateException("group set already");
        }
        //bossgroup 赋值给了AbstractBootstrap的group
        this.group = group;
        return self();
    }

2、channel方法-构建NioServerSocketChannel的实例工厂ReflectiveChannelFactory并赋值给ServerBootstrap的属性

    public B channel(Class<? extends C> channelClass) {
        //构建ReflectiveChannelFactory工厂实例
        return channelFactory(new ReflectiveChannelFactory<C>(
                ObjectUtil.checkNotNull(channelClass, "channelClass")
        ));
    }
    public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {
        
        private final Constructor<? extends T> constructor;
        
        //new 的时候会把传进来的class也就是NioServerSocketChannel的构造器赋值给constructor
        public ReflectiveChannelFactory(Class<? extends T> clazz) {
            ObjectUtil.checkNotNull(clazz, "clazz");
            try {
                this.constructor = clazz.getConstructor();
            } catch (NoSuchMethodException e) {
                throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                        " does not have a public non-arg constructor", e);
            }
        }
        
        //调用工厂的newChannel方法就可以通过构造器，反射得到class的实例对象
        @Override
        public T newChannel() {
            try {
                return constructor.newInstance();
            } catch (Throwable t) {
                throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
            }
        }

    }
    
    //将channel工厂实例赋值给channelFactory属性备用
    public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) {
        return channelFactory((ChannelFactory<C>) channelFactory);
    }
    public B channelFactory(ChannelFactory<? extends C> channelFactory) {
        ObjectUtil.checkNotNull(channelFactory, "channelFactory");
        if (this.channelFactory != null) {
            throw new IllegalStateException("channelFactory set already");
        }

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

3、handler方法-属性赋值

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

4、childHandler方法-属性赋值

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

到这里都是一些赋值操作，为真正的启动做准备。

二、ServerBootstrap.bind()

    //创建封装端口的对象InetSocketAddress
    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }

    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
    }
    //校验group和channelFactory
    public B validate() {
        if (group == null) {
            throw new IllegalStateException("group not set");
        }
        if (channelFactory == null) {
            throw new IllegalStateException("channel or channelFactory not set");
        }
        return self();
    }

    private ChannelFuture doBind(final SocketAddress localAddress) {
        // 负责NioServerSocketChannel和ServerSocketChannel的创建
        // ServerSocketChannel的注册工作
        // init由main线程完成，regisetr由NIO线程完成
        final ChannelFuture regFuture = initAndRegister();
        
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
        
        // 因为register操作是异步的
        // 所以要判断主线程执行到这里时，register操作是否已经执行完毕
        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            // 执行doBind0绑定操作
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            // Registration future is almost always fulfilled already, but just in case it's not.
            // 如果register操作还没执行完，就会到这个分支中来
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            // 添加监听器，NIO线程异步进行doBind0操作
            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;
        }
    }

• doBind()中有两个重要方法initAndRegister()和doBind0(regFuture, channel, localAddress, promise)

• initAndRegister主要负责NioServerSocketChannel和ServerSocketChannel的创建（主线程中完成）与ServerSocketChannel注册（NIO线程中完成）工作

• doBind0则负责连接的创建工作

1.1、initAndRegister()

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            //channelFactory即ReflectiveChannelFactory，调用newChannel()通过反射得到一个NioServerSocketChannel实体对象
            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;
    }
    
    @Override
    void init(Channel channel) {
        setChannelOptions(channel, newOptionsArray(), logger);
        setAttributes(channel, attrs0().entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY));
        //获取channel的对应的管道
        ChannelPipeline p = channel.pipeline();

        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(EMPTY_OPTION_ARRAY);
        }
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs = childAttrs.entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY);
        // 向Pipeline中添加了一个handler，该handler等待被调用
        p.addLast(new ChannelInitializer<Channel>() {
　　　　　　　//register之后才会调用
            @Override
            public void initChannel(final Channel ch) {
                final ChannelPipeline pipeline = ch.pipeline();
　　　　　　　　　 //此handler就是示例代码中添加的handler
                ChannelHandler handler = config.handler();
                if (handler != null) {
　　　　　　　　　　　　//添加到pipeline中
                    pipeline.addLast(handler);
                }
　　　　　　　　　 //添加新任务，任务负责添加handler，该handler负责发生Accepet事件后建立连接
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        //添加handler，在发生Accept事件后建立连接 
                        pipeline.addLast(new ServerBootstrapAcceptor(
　　　　　　　　　　　　　　　　　　　//currentChildHandler就是示例代码中childHandler
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

1.1.1、channelFactory.newChannel()之前说过这个地方是利用反射调用NioServerSocketChannel的构造器创建实例，下面来分析以下实例化过程：

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

    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();

    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>.
             */
            //通过SelectorProvider打开ServerSocketChannel并返回
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException(
                    "Failed to open a server socket.", e);
        }
    }

    private final ServerSocketChannelConfig config;

    //反射调用的默认构造器
    public NioServerSocketChannel() {
        //newSocket就是在创建原始的ServerSocketChannel
        //调用this构造器
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }

    public NioServerSocketChannel(ServerSocketChannel channel) {
        //调用父类构造器，其中的一个目的就是赋值属性
        super(null, channel, SelectionKey.OP_ACCEPT);
        //初始化存储配置类
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
}
//父类
public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
    protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent, ch, readInterestOp);
    }
}
//父类
public abstract class AbstractNioChannel extends AbstractChannel {

    private final SelectableChannel ch;
    protected final int readInterestOp;
    volatile SelectionKey selectionKey;
    boolean readPending;

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        //将ServerSocketChannel赋值给ch属性
        this.ch = ch;
        //将SelectionKey.OP_ACCEPT事件赋值给readInterestOp属性
        this.readInterestOp = readInterestOp;
        try {
            //设置ServerSocketChannel非阻塞
            ch.configureBlocking(false);
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
                logger.warn(
                            "Failed to close a partially initialized socket.", e2);
            }

            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }
}

 总结：init方法主要完成了以下三个操作

　　1）创建NioServerSocketChannel
　　2）通过NioServerSocketChannel的构造器，创建了ServerSocketChannel
　　3）由initChannel方法向NioServerSocketChannel中添加了两个handler，添加操作在register之后被执行
　　　　一个handler负责设置配置
　　　　一个handler负责发生Accepet事件后建立连接

1.2、ChannelFuture regFuture = config().group().register(channel)操作，config().group()获取的是bossGroup，是一个NioEventLoopGroup继承了MultithreadEventLoopGroup，register方法就是MultithreadEventLoopGroup的一个方法，具体看下面源码：

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

    @Override
    public ChannelFuture register(Channel channel) {
        //封装了channel和线程执行器
        return register(new DefaultChannelPromise(channel, this));
    }

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

        @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;
            }
            if (!isCompatible(eventLoop)) {
                promise.setFailure(
                        new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
                return;
            }
            // 获取EventLoop
            AbstractChannel.this.eventLoop = eventLoop;
            // 此处完成了由 主线程 到 NIO线程 的切换
            // eventLoop.inEventLoop()用于判断当前线程是否为NIO线程
            if (eventLoop.inEventLoop()) {
                register0(promise);
            } else {
                try {
                    // 向NIO线程中添加任务
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            // 该方法中会执行doRegister
                            // 执行真正的注册操作
                            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);
                }
            }
        }

        private void register0(ChannelPromise promise) {
            try {
                // check if the channel is still open as it could be closed in the mean time when the register
                // call was outside of the eventLoop
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                boolean firstRegistration = neverRegistered;
                // 执行真正的注册操作
                doRegister();
                neverRegistered = false;
                registered = true;

                // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
                // user may already fire events through the pipeline in the ChannelFutureListener.
                // 调用init中的initChannel方法
                pipeline.invokeHandlerAddedIfNeeded();

                safeSetSuccess(promise);
                pipeline.fireChannelRegistered();
                // Only fire a channelActive if the channel has never been registered. This prevents firing
                // multiple channel actives if the channel is deregistered and re-registered.
                if (isActive()) {
                    if (firstRegistration) {
                        pipeline.fireChannelActive();
                    } else if (config().isAutoRead()) {
                        // This channel was registered before and autoRead() is set. This means we need to begin read
                        // again so that we process inbound data.
                        //
                        // See https://github.com/netty/netty/issues/4805
                        //关注Accept事件
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                // Close the channel directly to avoid FD leak.
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }        
//AbstractNioChannel        
@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            // javaChannel()即为ServerSocketChannel
            // eventLoop().unwrappedSelector()获取eventLoop中的Selector
            // this为NIOServerSocketChannel，作为附件
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            ...
           
        }
    }
}

register方法主要完成了以下三个操作：
1）完成了主线程到NIO的线程切换
　　通过eventLoop.inEventLoop()进行线程判断，判断当前线程是否为NIO线程
　　切换的方式为让eventLoop执行register的操作
　　register的操作在NIO线程中完成
2）调用doRegister方法
　　selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
　　将ServerSocketChannel注册到EventLoop的Selector中
　　此时还未关注事件
　　添加NioServerSocketChannel附件
3）通过invokeHandlerAddedIfNeeded调用init中的initChannel方法
　　initChannel方法主要创建了两个handler
　　一个handler负责设置配置
　　一个handler负责发生Accept事件后建立连接

2、doBind0()方法

　　在doRegister和invokeHandlerAddedIfNeeded操作中的完成后，会调用safeSetSuccess(promise)方法，向Promise中设置执行成功的结果。此时doBind方法中由initAndRegister返回的ChannelFuture对象regFuture便会由NIO线程异步执行doBind0绑定操作。

2.1、doBind0最底层调用的是ServerSocketChannel的bind方法，通过该方法，绑定了对应的端口。

    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());
                }
            }
        });
    }
    //AbstractChannel
    @Override
    public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
        return pipeline.bind(localAddress, promise);
    }

    //DefaultChannelPipeline
    @Override
    public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
        return tail.bind(localAddress, promise);
    }
    
    //AbstractChannelHandlerContext
    @Override
    public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
        ObjectUtil.checkNotNull(localAddress, "localAddress");
        if (isNotValidPromise(promise, false)) {
            // cancelled
            return promise;
        }

        final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
        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, false);
        }
        return promise;
    }
    
    //AbstractChannelHandlerContext
    private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
        if (invokeHandler()) {
            try {
                ((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
            } catch (Throwable t) {
                notifyOutboundHandlerException(t, promise);
            }
        } else {
            bind(localAddress, promise);
        }
    }
        
        //DefaultChannelPipeline HeadContext
        @Override
        public void bind(
                ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
            unsafe.bind(localAddress, promise);
        }
        
        //AbstractChannel 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;
            }

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

            safeSetSuccess(promise);
        }
    //NioServerSocketChannel    
    @SuppressJava6Requirement(reason = "Usage guarded by java version check")
    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

2.2、beginRead()方法

在绑定端口操作完成后，会判断各种所有初始化操作是否已经完成，若完成，则会添加ServerSocketChannel感兴趣的事件

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

最终在AbstractNioChannel.doBeginRead方法中，会添加Accept事件

@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();
    // 如果ServerSocketChannel没有关注Accept事件
    if ((interestOps & readInterestOp) == 0) {
        // 则让其关注Accepet事件
        // readInterestOp 取值是 16
        // 在 NioServerSocketChannel 创建时初始化
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

总结：SererBootStrap的初始化及bind方法主要做了以下的事情

　　1）NioServerSocketChannel与ServerSocketChannel的创建
　　2）ServerSocketChannel绑定到EventLoop的Selecot中，并添加NioServerSocketChannel附件　　
　　3）绑定了对应的端口
　　4）关注了Accept事件

 三、启动流程简化梳理

// netty 中使用 NioEventLoopGroup （简称 nio boss 线程）来封装线程和 selector
Selector selector = Selector.open(); 

// 创建 NioServerSocketChannel，同时会初始化它关联的 handler，以及为原生 ssc 存储 config
NioServerSocketChannel attachment = new NioServerSocketChannel();

// 创建 NioServerSocketChannel 时，创建了 java 原生的 ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); 
serverSocketChannel.configureBlocking(false);

// 启动 nio boss 线程执行接下来的操作

//注册（仅关联 selector 和 NioServerSocketChannel），未关注事件
SelectionKey selectionKey = serverSocketChannel.register(selector, 0, attachment);

// head -> 初始化器 -> ServerBootstrapAcceptor -> tail，初始化器是一次性的，只为添加 acceptor

// 绑定端口
serverSocketChannel.bind(new InetSocketAddress(8080));

// 触发 channel active 事件，在 head 中关注 op_accept 事件
selectionKey.interestOps(SelectionKey.OP_ACCEPT);

 参考链接：https://nyimac.gitee.io/2021/04/25/Netty%E5%9F%BA%E7%A1%80/