netty的引导程序ServerBootStrap

墨蓝 2022-07-13 01:39 168阅读 0赞

BootStrap在netty的应用程序中负责引导服务器和客户端。netty包含了两种不同类型的引导：  
1. 使用服务器的ServerBootStrap，用于接受客户端的连接以及为已接受的连接创建子通道。  
2. 用于客户端的BootStrap，不接受新的连接，并且是在父通道类完成一些操作。

# ServerBootStrap的运行原理 #

下面先看一下这两个引导类的类继承图：  
服务端的ServerBootstrap类继承图：  
![服务端的ServerBootstrap类继承图][ServerBootstrap]

客户端的Bootstrap类继承图：  
![客户端的Bootstrap类继承图][Bootstrap]

下面我们分析一下服务端的ServerBootstrap的源码来分析引导流程：  
首先给出一个很简单的基于netty的聊天室的服务端的实例：

package netty.cookbook.simplechat;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import utils.LogUtil;
    
    /** * Created by louyuting on 16/12/8. * 启动服务端 */
    public class SimpleChatServer { 
        private int port;
    
        public SimpleChatServer(int port){
            this.port = port;
        }
    
        public void run() throws Exception{
            //NioEventLoopGroup是用来处理IO操作的多线程事件循环器
            //boss用来接收进来的连接
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            //用来处理已经被接收的连接;
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try{
                //是一个启动NIO服务的辅助启动类
                ServerBootstrap sBootstrap = new ServerBootstrap();
                //These EventLoopGroup's are used to handle all the events and IO for ServerChannel and Channel's.
                //为bootstrap设置acceptor的EventLoopGroup和client的EventLoopGroup
                //这些EventLoopGroups用于处理所有的IO事件
                //?这里为什么设置两个group呢?
                sBootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new SimpleChatServerInitializer())
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
    
                LogUtil.log_debug("SimpleChatServer 启动了");
    
                //绑定端口,开始接收进来的连接
                ChannelFuture future = sBootstrap.bind(port).sync();
                //等待服务器socket关闭
                //在本例子中不会发生,这时可以关闭服务器了
                future.channel().closeFuture().sync();
            } finally {
                //
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
    
                LogUtil.log_debug("SimpleChatServer 关闭了");
            }
    
        }
    
        public static void main(String[] args) throws Exception {
            new SimpleChatServer(8080).run();
        }
    }

下面以上面的实例来分析ServerBootStrap的运行流程：

### 1.实例化ServerBootstrap ###

`ServerBootstrap sBootstrap = new ServerBootstrap();`  
首先就是通过new关键字实例化一个ServerBootStrap对象。

### 2.配置ServerBootstrap的group() ###

//boss用来接收进来的连接
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    //用来处理已经被接收的连接;
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    sBootstrap.group(bossGroup, workerGroup)
              .channel(NioServerSocketChannel.class)
              .childHandler(new SimpleChatServerInitializer())
              .option(ChannelOption.SO_BACKLOG, 128)
              .childOption(ChannelOption.SO_KEEPALIVE, true);

从上面可知，上面创建了两个EventLoopGroup，分别是boss和worker，然后配置到ServerBootstrap的group中。我们先来看看ServerBootstrap.group()，这个函数有两个重载的实现：

public ServerBootstrap group(EventLoopGroup group) {
        return group(group, group);
    }
    
    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        super.group(parentGroup);
        if (childGroup == null) {
            throw new NullPointerException("childGroup");
        }
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        }
        this.childGroup = childGroup;
        return this;
    }

可知，这里如果调用时，如果只传入了一个EventLoopGroup，最后也会调用group(EventLoopGroup parentGroup, EventLoopGroup childGroup)。  
这里传入的两个EventLoopGroup分别叫做parentGroup和childGroup。其实我觉得更加好理解的方式应该叫boss和worker。boss这个EventLoopGroup作为一个acceptor负责接收来自客户端的请求，然后分发给worker这个EventLoopGroup来处理所有的事件event和channel的IO。  
查看上面的源码，我们可知，首先调用的是  
`super.group(parentGroup);`  
这个方法调用了ServerBootstrap的父类AbstractBootstrap的group(EventLoopGroup group) 方法，下面看看AbstractBootstrap中方法的定义：

public B group(EventLoopGroup group) {
        //传参不能为空
        if (group == null) {
            throw new NullPointerException("group");
        }
        //AbstractBootstrap的group属性如果非空，就抛出异常，说明this.group是单例的
        if (this.group != null) {
            throw new IllegalStateException("group set already");
        }
        this.group = group;
        return (B) this;
    }

从上面的源码AbstractBootstrap里面的这个this.group属性是单例的。我们先看看该属性的定义：  
`volatile EventLoopGroup group;`  
这里定义的是volatile的变量，提供可见性，而且变量是包级别的权限。最后将传入的EventLoopGroup赋值给AbstractBootstrap的group属性。

后面的ServerBootstrap的childGroup也是同样的实现，不过childGroup是单例的赋值给了ServerBootstrap的childGroup属性。ServerBootstrap中的定义如下：  
`private volatile EventLoopGroup childGroup;`

### 3.配置ServerBootstrap的channel() ###

这里以聊天室的实例为例，  
`ServerBootstrap.channel(NioServerSocketChannel.class)`  
实际上是调用的是AbstractBootstrap里面的channel()函数，下面先粘贴出该方法的源码：

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

这里传入的是一个Class对象，根据传入的不同的Class对象，实例化不同的Channel，主要是有两种代表NIO和OIO的对象：NioServerSocketChannel和OioServerSocketChannel。在函数体的最后调用了：  
`channelFactory(new ReflectiveChannelFactory<C>(channelClass));`  
参数里面实例化了一个 ReflectiveChannelFactory对象，这个对象实现了ChannelFactory这个接口的。  
下面进入这个函数,最后调用的是：

public B channelFactory(ChannelFactory<? extends C> channelFactory) {
        //判断传参非空
        if (channelFactory == null) {
            throw new NullPointerException("channelFactory");
        }
        //AbstractBootstrap的channelFactory属性非空
        if (this.channelFactory != null) {
            throw new IllegalStateException("channelFactory set already");
        }
        //传递给channelFactory属性
        this.channelFactory = channelFactory;
        return (B) this;
    }

可知，最后是把new的ReflectiveChannelFactory传递给了AbstractBootstrap的channelFactory属性，该属性定义如下：  
`private volatile ChannelFactory<? extends C> channelFactory;`

### 4.配置ServerBootstrap的childHandler(ChannelHandler childHandler) ###

该函数的主要作用是设置channelHandler来处理客户端的请求的channel的IO。  
这里我们一般都用ChannelInitializer这个类的实例或则继承自这个类的实例，在我的聊天室程序中实例如下：  
`ServerBootstrap.childHandler(new SimpleChatServerInitializer())`  
这里我是通过新建类SimpleChatServerInitializer继承自ChannelInitializer。具体的代码如下：

/** * Created by louyuting on 16/12/8. * 用来增加多个的处理类到ChannelPipeline上:包括编码,解码,SimpleChatServerHandler */
    public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel>{ 
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
    
            pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
            pipeline.addLast("decoder", new StringDecoder());
            pipeline.addLast("encoder", new StringEncoder());
            pipeline.addLast("handler", new SimpleChatServerHandler());
    
            System.out.println("SimpleChatClient:" + ch.remoteAddress()+"连接上");
        }
    }

我们再看看ChannelInitializer这个类的继承图可知ChannelInitializer其实就是继承自ChannelHandler的  
![ChannelInitializer类图][ChannelInitializer]  
可知，这个类其实就是往pipeline中添加了很多的channelHandler。在ServerBootstrap.childHandler(ChannelHandler childHandler)中的实现是：

public ServerBootstrap childHandler(ChannelHandler childHandler) {
        if (childHandler == null) {
            throw new NullPointerException("childHandler");
        }
        this.childHandler = childHandler;
        return this;
    }

由最后一句可知，其实就是讲传入的childHandler赋值给ServerBootstrap的childHandler属性。

### 5.配置ServerBootstrap的option(ChannelOption option, T value) ###

这里调用的是父类的AbstractBootstrap的option()方法，源码如下：

public <T> B option(ChannelOption<T> option, T value) {
        if (option == null) {
            throw new NullPointerException("option");
        }
        if (value == null) {
            synchronized (options) {
                options.remove(option);
            }
        } else {
            synchronized (options) {
                options.put(option, value);
            }
        }
        return (B) this;
    }

其中最重要的一行代码就是：  
`options.put(option, value);`  
这里用到了options这个参数，在AbstractBootstrap的定义如下：  
`private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();`  
可知是私有变量，而且是一个Map集合。这个变量主要是设置TCP连接中的一些可选项,而且这些属性是作用于每一个连接到服务器被创建的channel。

### 6.配置ServerBootstrap的childOption(ChannelOption childOption, T value) ###

这里调用的是父类的ServerBootstrap的childOption()方法，源码如下：

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

这个函数功能与option()函数几乎一样，唯一的区别是该属性设定只作用于被acceptor(也就是boss EventLoopGroup)接收之后的channel。

### 7.配置ServerBootstrap的绑定端口和启动： ###

上面介绍的都是关于ServerBootstrap的一些配置的绑定，下面介绍在启动服务器时候做的工作：

通过聊天室的服务端程序我们可以知道，绑定端口并启动服务程序如下：

//绑定端口,开始接收进来的连接
    ChannelFuture future = sBootstrap.bind(port).sync();

bind()函数是AbstractBootstrap里面的方法，首先分析bind(port)函数的功能：直接看源码：

/** * 1. *创建一个通道并绑定到当前BootStrap */
    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }
    
    /** * 2. *创建一个通道并绑定到当前BootStrap */
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();//
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }

从上面的源码可知bind(port)最后会调用bind(SocketAddress localAddress)函数，里面的 validate()函数会先校验AbstractBootstrap的成员属性group(也就是boss)和channelFactory非空。

最后就是调用doBind(localAddress);方法了，这里才是bind()函数的核心：看源码分析这个函数做了什么工作：

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

从定义可知，这个函数是一个private函数，也就是类内部调用的。首先看  
`final ChannelFuture regFuture = initAndRegister();`

里面initAndRegister()函数的功能：

final ChannelFuture initAndRegister() {
        //new 了一个新的channel
        Channel channel = null;
        try {
            //调用了AbstractBootstrap的channelFactory属性新建了一个在ServerBootStrap中配置的channel类型，在我的聊天室程序中是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);
        }
    
        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;
    }

在initAndRegister()函数中，首先调用了AbstractBootstrap的channelFactory.newChannel()方法新建了一个在ServerBootStrap中配置的channel类型，在我的聊天室程序中NioServerSocketChannel。

然后调用init()函数初始化这个通道(该函数在ServerBootstrap中实现)：完整的源码就不给出的，主要说说里面做了什么任务：

#### (1).获取ServerBootStrap中options和attrs的配置，然后设置在新创建的channel对象中，代码如下： ####

final Map<ChannelOption<?>, Object> options = options0();
    synchronized (options) {
        channel.config().setOptions(options);
    }
    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());
        }
    }

#### (2).获取新new的channel的pipeline，然后添加handler到channel中。具体看源码注释 ####

//获取channel的pipeline
    ChannelPipeline p = channel.pipeline();
    
    /** currentChildGroup也就是worker的EventLoopGroup*/
    final EventLoopGroup currentChildGroup = childGroup;
    /**currentChildHandler也就是在ServerBootStrap中添加的childHandler*/
    final ChannelHandler currentChildHandler = childHandler;
    /** 获取worker角色的EventLoopGroup的option设置和attr设置并配置到currentChildOptions和currentChildAttrs中 */
    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()));
    }
    
    /** 向new的channel中添加handler **/
    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            //获取ServerBootStrap中添加的childhandler，然后添加到channel中
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }
    
            // We add this handler via the EventLoop as the user may have used a ChannelInitializer as handler.
            // In this case the initChannel(...) method will only be called after this method returns. Because
            // of this we need to ensure we add our handler in a delayed fashion so all the users handler are
            // placed in front of the ServerBootstrapAcceptor.
            //新建一个线程，将channelHandler的子类之new的ServerBootstrapAcceptor添加到pipeline中。
            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    pipeline.addLast(new ServerBootstrapAcceptor(
                            currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });

下面再看看ServerBootstrapAcceptor这个类里面做了什么？ServerBootstrapAcceptor是ServerBootstrap里面的静态类，这个类本身也是一个channelHandler，我们分析它的channelRead(ChannelHandlerContext ctx, Object msg)方法,主要还是做了以下几个工作：  
1. 向channel中添加handler；  
2. 向worker这个EventLoopGroup中注册当前channel

分析完init()函数，我们继续看initAndRegister()函数中的其余部分：  
接下来往boss 这个EventLoopGroup中注册当前channel。

最后在doBind()函数中，都会调用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());
                }
            }
        });
    }

从上面的源码可知，最后都会调用channel.bind()函数。

[ServerBootstrap]: /images/20220713/22fea1e9ad3743859aa33c5abcc45276.png
[Bootstrap]: /images/20220713/95463e7db8d6491d995288950cf5fd4f.png
[ChannelInitializer]: /images/20220713/15c27acf16c841b3a1b820c30ba06b7c.png