Netty源码分析 - Bootstrap服务端

系列

Netty源码分析 - Bootstrap服务端
Netty源码分析 - Bootstrap客户端
netty源码分析 - ChannelHandler
netty源码分析 - EventLoop类关系
netty源码分析 - register分析
Netty源码分析 - NioEventLoop事件处理
netty源码分析 - accept过程分析
Netty源码分析 - ByteBuf
Netty源码分析 - 粘包和拆包问题

开篇

• 本文基于netty-4.1.8.Final版本进行分析，主要是分析Netty Server初始化过程。
• 建议先参考Netty源码分析 - Bootstrap客户端文章。
• 理解NioServerSocketChannel的初始化和注册流程。
• 理解ServerBootstrap的bind流程，理解bossGroup与workerGroup的关系。
  
  

Netty Server案例

public final class DiscardServer {

    static final boolean SSL = System.getProperty("ssl") != null;
    static final int PORT = Integer.parseInt(System.getProperty("port", "8009"));

    public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }
        // 父bossGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 子workerGroup
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 创建ServerBootstrap
            ServerBootstrap b = new ServerBootstrap();
            // 绑定group
            b.group(bossGroup, workerGroup)
              // 绑定channel
             .channel(NioServerSocketChannel.class)
              // 绑定父handler
             .handler(new LoggingHandler(LogLevel.INFO))
              // 绑定子handler
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                      // 添加handler到ChannelPipeline
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     p.addLast(new DiscardServerHandler());
                 }
             });

            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            // In this example, this does not happen, but you can do that to gracefully
            // shut down your server.
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

ServerBootstrap

ServerBootstrap

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(ServerBootstrap.class);

    private final Map<ChannelOption<?>, Object> childOptions = new LinkedHashMap<ChannelOption<?>, Object>();
    private final Map<AttributeKey<?>, Object> childAttrs = new LinkedHashMap<AttributeKey<?>, Object>();
    private final ServerBootstrapConfig config = new ServerBootstrapConfig(this);
    private volatile EventLoopGroup childGroup;
    private volatile ChannelHandler childHandler;

    public ServerBootstrap() { }

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

    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        super.group(parentGroup);
        this.childGroup = childGroup;
        return this;
    }

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

    public <T> ServerBootstrap childAttr(AttributeKey<T> childKey, T value) {
        if (value == null) {
            childAttrs.remove(childKey);
        } else {
            childAttrs.put(childKey, value);
        }
        return this;
    }

    public ServerBootstrap childHandler(ChannelHandler childHandler) {
        this.childHandler = childHandler;
        return this;
    }

    @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(childOptions.size()));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
        }

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(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(
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

• ServerBootstrap和Bootstrap的区别在于有child相关的子属性，包括childGroup、childHandler、childOptions、childAttrs。
• ServerBootstrap和Bootstrap的区别在于有child相关的方法，包括childGroup()、childHandler()、childOptions()、childAttrs()。
• ServerBootstrap的childGroup用于保存workerGroup对象，父类AbstractBootstrap用于保存bossGroup对象。
• ServerBootstrap重载了父类AbstractBootstrap的group()方法，在不区分bossGroup和workerGroup的场景下，共用bossGroup。
• ServerBootstrap重载了父类的initChannel的方法init()，这个很关键，起到了连接bossGroup和workerGroup的作用，通过ServerBootstrapAcceptor起作用。
  
  

NioEventLoopGroup/NioEventLoop

• NioEventLoopGroup/NioEventLoop和《Netty源码分析 - Bootstrap客户端》同章节内容相同，可以参考Netty源码分析 - Bootstrap客户端同部分的内容。

NioServerSocketChannel

NioServerSocketChannel

public class NioServerSocketChannel extends AbstractNioMessageChannel
                             implements io.netty.channel.socket.ServerSocketChannel {
    private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
    // SelectorProvider.provider()返回provider单例对象
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    private final ServerSocketChannelConfig config;

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NioServerSocketChannel.class);

    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openServerSocketChannel();
        } catch (IOException e) {
        }
    }

    public NioServerSocketChannel() {
        // newSocket()返回ServerSocketChannel对象
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }

    public NioServerSocketChannel(ServerSocketChannel channel) {
        // 关注SelectionKey.OP_ACCEPT事件
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
}


public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
    boolean inputShutdown;

    protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent, ch, readInterestOp);
    }

    protected AbstractNioUnsafe newUnsafe() {
        // 创建的是NioMessageUnsafe对象
        return new NioMessageUnsafe();
    }
}


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);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false);
        } catch (IOException e) {
        }
    }
}


public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    private final Channel parent;
    private final ChannelId id;
    private final Unsafe unsafe;
    private final DefaultChannelPipeline pipeline;
    private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);
    private final CloseFuture closeFuture = new CloseFuture(this);
    private volatile EventLoop eventLoop;
    private volatile boolean registered;

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

• NioServerSocketChannel初始化按照子类到父类的顺序，NioServerSocketChannel => AbstractNioMessageChannel => AbstractNioChannel => AbstractChannel。
• NioServerSocketChannel和NioSocketChannel的大部分变量都是相同的，有区别的变量在于NioServerSocketChannelConfig的config对象、NioMessageUnsafe的unsafe对象。
• NioServerSocketChannel调用父类AbstractNioMessageChannel构造器时, 传入的参数是 SelectionKey.OP_ACCEPT。作为对比 在客户端的 Channel 初始化时, 传入的参数是 SelectionKey.OP_READ。

NioMessageUnsafe

NioMessageUnsafe

public abstract class AbstractNioMessageChannel extends AbstractNioChannel {

    protected AbstractNioUnsafe newUnsafe() {
        return new NioMessageUnsafe();
    }

    private final class NioMessageUnsafe extends AbstractNioUnsafe {

        private final List<Object> readBuf = new ArrayList<Object>();

        @Override
        public void read() {
            assert eventLoop().inEventLoop();
            final ChannelConfig config = config();
            final ChannelPipeline pipeline = pipeline();
            final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
            allocHandle.reset(config);

            boolean closed = false;
            Throwable exception = null;
            
            // 省略相关代码            
        }
    }
}

• NioMessageUnsafe是NioServerSocketChannel特有的unsafe对象。
• NioMessageUnsafe理论上是在server端处理读写连接事件的对象。

NioServerSocketChannel使用流程

• 参考Netty源码分析 - Bootstrap客户端《NioSocketChannel初始化注册过程》章节内容，除部分内容不同除外。

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

    private ChannelFuture doBind(final SocketAddress localAddress) {
        // 初始化Channel流程，通过initAndRegister()执行
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }

        // 省略相关代码
    }

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            // 1.创建channel对象，指的是NioServerSocketChannel类
            channel = channelFactory.newChannel();
            // 2.初始化channel对象，在ServerBootstrap类重写了init()方法
            init(channel);
        } catch (Throwable t) {
            // 省略相关代码
        }
        // 3.注册channel对象
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }

        return regFuture;
    }
}

• AbstractBootstrap#initAndRegister过程中channelFactory.newChannel()创建的是NioServerSocketChannel对象。
• AbstractBootstrap#initAndRegister过程中init(channel)初始化NioServerSocketChannel对象，init()方法在ServerBootstrap#init()被重载。
• config().group().register(channel)执行过程和Netty源码分析 - Bootstrap客户端《NioSocketChannel初始化注册过程》相同，group()返回的是bossGroup。

Bootstrap bind流程

bind流程

在 NioServerSocketChannel 实例化过程中, 所需要做的工作:

• 调用 NioServerSocketChannel.newSocket(DEFAULT_SELECTOR_PROVIDER) 打开一个新的 Java NIO ServerSocketChannel

• AbstractNioChannel 中的属性:
  SelectableChannel ch 被设置为 Java ServerSocketChannel, 即 NioServerSocketChannel#newSocket 返回的 Java NIO ServerSocketChannel
  readInterestOp 被设置为 SelectionKey.OP_ACCEPT
  SelectableChannel ch 被配置为非阻塞的 ch.configureBlocking(false)

• AbstractChannel(Channel parent) 中初始化 AbstractChannel 的属性:
  parent 属性置为 null
  unsafe 通过newUnsafe() 实例化一个 unsafe 对象, 它的类型是 AbstractNioMessageChannel#AbstractNioUnsafe 内部类
  pipeline 是 new DefaultChannelPipeline(this) 新创建的实例

• NioServerSocketChannel 中的属性:
  ServerSocketChannelConfig config = new NioServerSocketChannelConfig(this, javaChannel().socket())

在 NioServerSocketChannel bind过程中, 所需要做的工作:

• 按照 NioServerSocketChannel#bind => DefaultChannelPipeline#bind => TailContext#bind =>HeadContext#bind => NioMessageUnsafe#bind => NioServerSocketChannel#doBind顺序进行绑定。

bossGroup与workerGroup

bossGroup&workerGroup

• 首先, 服务器端 bossGroup 不断地监听是否有客户端的连接, 当发现有一个新的客户端连接到来时, bossGroup 就会为此连接初始化各项资源, 然后从 workerGroup 中选出一个 EventLoop 绑定到此客户端连接中. 那么接下来的服务器与客户端的交互过程就全部在此分配的 EventLoop 中了。

服务器端的handler添加流程

• 在服务器 NioServerSocketChannel 的 pipeline 中添加的是 handler 与 ServerBootstrapAcceptor
• 当有新的客户端连接请求时, ServerBootstrapAcceptor.channelRead 中负责新建此连接的 NioSocketChannel 并添加 childHandler 到 NioSocketChannel 对应的 pipeline 中, 并将此 channel 绑定到 workerGroup 中的某个 eventLoop 中
• handler 是在 accept 阶段起作用, 它处理客户端的连接请求
• childHandler 是在客户端连接建立以后起作用, 它负责客户端连接的 IO 交互.

bossGroup与workerGroup源码分析

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {
    @Override
    void init(Channel channel) throws Exception {

        // 省略相关代码

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(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() {
                        // NioServerSocketChannel绑定ServerBootstrapAcceptor
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }


    private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {

        private final EventLoopGroup childGroup;
        private final ChannelHandler childHandler;
        private final Entry<ChannelOption<?>, Object>[] childOptions;
        private final Entry<AttributeKey<?>, Object>[] childAttrs;

        ServerBootstrapAcceptor(
                EventLoopGroup childGroup, ChannelHandler childHandler,
                Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
            this.childGroup = childGroup;
            this.childHandler = childHandler;
            this.childOptions = childOptions;
            this.childAttrs = childAttrs;
        }

        @Override
        @SuppressWarnings("unchecked")
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            final Channel child = (Channel) msg;
            // // NioSocketChannel绑定childHandler
            child.pipeline().addLast(childHandler);

            setChannelOptions(child, childOptions, logger);

            for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
            }

            try {
                childGroup.register(child).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            forceClose(child, future.cause());
                        }
                    }
                });
            } catch (Throwable t) {
                forceClose(child, t);
            }
        }
    }
}

• ServerBootstrap#init方法中NioServerSocketChannel对应的pipeline会绑定ServerBootstrapAcceptor的handler，新建连接会通过该handler进行处理。
• ServerBootstrapAcceptor#channelRead针对新建的连接通过child.pipeline().addLast(childHandler)绑定handler，同时绑定到childGroup的NioEventLoop当中。

ServerBootstrapAcceptor

• ServerBootstrapAcceptor属于ChannelHandler对象。

参考

Netty 源码分析之 一 揭开 Bootstrap 神秘的红盖头 (服务器端)