Netty 核心源码解读 —— ServerBootstrap 篇

原创张松然松然聊技术

本文我们就开始对 ServerBootstrap 进行源码解读（4.1.51.Final-SNAPSHOT），为什么是 ServerBootstrap，记得在用 Netty 做第一个项目的时候，写的第一行 Code 就是 new ServerBootstrap()，ServerBootstrap 是 Netty Server 的启动类，所以从它开始了解 Netty 是最合适的。

ServerBootstrap

## TcpServer.java
private final EventLoopGroup bossGroup = new NioEventLoopGroup(BIZ_GROUP_SIZE);private final EventLoopGroup workerGroup = new NioEventLoopGroup(BIZ_THREAD_SIZE);
public void init() throws Exception {    // Server 服务启动    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup);    bootstrap.channel(NioServerSocketChannel.class);    bootstrap.childHandler(new ServerChannelInitializer(serverConfig));    // 可选参数    bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
    // 绑定接口，同步等待成功    ChannelFuture future = bootstrap.bind(port).sync();    ChannelFuture channelFuture = future.addListener(new ChannelFutureListener() {        public void operationComplete(ChannelFuture future) throws Exception {        }    });}

这是我在做 TCP 网关时写的 Netty Server 的代码片段（https://github.com/SongranZhang/tcp-gateway/blob/master/src/main/java/com/linkedkeeper/tcp/connector/tcp/server/TcpServer.java），可以看到，Netty Server 的初始化首先是通过 ServerBootstrap 的无参构造函数创建一个对象，接着是这个对象的一串链式调用 bootstrap.group().channel().childHandler().childOption()，而服务启动的真正触发点是这段 bootstrap.bind(port).sync()，下面我们就逐一来分析下这里的每个方法。

首先是 group() 方法。

## ServerBootstrap.java
private volatile EventLoopGroup childGroup;
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {    super.group(parentGroup);    if (this.childGroup != null) {        throw new IllegalStateException("childGroup set already");    }    this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");    return this;}

这里 workerGroup 赋值给了 ServerBootstrap 的 childGroup，bossGroup 赋值给了父类 AbstractBootstrap 的 group。

## AbstractBootstrap.java
volatile EventLoopGroup group;
public B group(EventLoopGroup group) {    ObjectUtil.checkNotNull(group, "group");    if (this.group != null) {        throw new IllegalStateException("group set already");    }    this.group = group;    return self();}

接下来是 channel() 方法。

## AbstractBootstrap.java
private volatile ChannelFactory<? extends C> channelFactory;
public B channel(Class<? extends C> channelClass) {    return channelFactory(new ReflectiveChannelFactory<C>(            ObjectUtil.checkNotNull(channelClass, "channelClass")    ));}
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();}

这里 NioServerSocketChannel.class 通过 ReflectiveChannelFactory 进行了实例化，然后赋值给了 AbstractBootstrap 的 channelFactory。

接下来是 childHandler() 方法。

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

这里是对 ServerBootstrap 的 childHandler 赋值。

最后是 childOption() 方法。

## ServerBootstrap.java
private final Map<ChannelOption<?>, Object> childOptions = new LinkedHashMap<ChannelOption<?>, Object>();
public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value) {    ObjectUtil.checkNotNull(childOption, "childOption");    synchronized (childOptions) {        if (value == null) {            childOptions.remove(childOption);        } else {            childOptions.put(childOption, value);        }    }    return this;}

这里 childOptions 维护了 TCP 的参数设置。

简言之 bootstrap.group().channel().childHandler().childOption() 就是在构建 Netty Server 的各种参数，下面再来看 bootstrap.bind(port).sync()。

首先是 bind() 方法。

## AbstractBootstrap.java
public ChannelFuture bind(int inetPort) {    return bind(new InetSocketAddress(inetPort));}
public ChannelFuture bind(SocketAddress localAddress) {    validate();    return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));}
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();}

这里的 validate() 方法对 AbstractBootstrap 的 group 和 channelFactory 进行非空校验，之后调用 doBind() 方法。

## AbstractBootstrap.java
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;    }}

首先看一下 initAndRegister() 方法。

## AbstractBootstrap.java
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;}

这里 channelFactory.newChannel() 调用的是 ReflectiveChannelFactory 的 newChannel 方法。

## ReflectiveChannelFactory
private final Constructor<? extends T> 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);    }}
public T newChannel() {    try {        return constructor.newInstance();    } catch (Throwable t) {        throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);    }}

这里 constructor.newInstance() 是 NioServerSocketChannel.class 的一个实例。得到 channel 后，调用 init(channel) 进行初始化，一是给 options 和 attrs 赋值，二是构建 pipeline。

## ServerBootstrap.java
void init(Channel channel) {    setChannelOptions(channel, newOptionsArray(), logger);    setAttributes(channel, attrs0().entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY));
    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);
    p.addLast(new ChannelInitializer<Channel>() {        @Override        public void initChannel(final Channel ch) {            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));                }            });        }    });}

回到 initAndRegister() 方法中，init(channel) 之后是 register(channel)，该方法在 NioEventLoopGroup 的父类 MultithreadEventLoopGroup 中实现，我们在解读 NioEventLoop 源码时再分析。

## MultithreadEventLoopGroup
public ChannelFuture register(Channel channel) {    return next().register(channel);}

看完 initAndRegister()，再回到 doBind() 接着看 doBind0()。

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

这里 regFuture.isSuccess() 会执行 channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);，否者执行 promise.setFailure(regFuture.cause());，这里的 promise 可以认为是一种特殊的 Future 对象。bind 是在 ChannelPipeline 里进行绑定的，我们在解读 ChannelPipeline 源码时再分析。

最后看一下 bootstrap.bind(serverPort).sync() 中的 sync()，bootstrap.bind(serverPort) 返回的是 ChannelFuture，所以 sync() 是调用 DefaultChannelPromise 的方法。

## DefaultChannelPromise
public ChannelPromise sync() throws InterruptedException {    super.sync();    return this;}

这里 super.sync(); 调用了父类的方法。

## DefaultPromise
public Promise<V> sync() throws InterruptedException {    await();    rethrowIfFailed();    return this;}
public Promise<V> await() throws InterruptedException {    if (isDone()) {        return this;    }
    if (Thread.interrupted()) {        throw new InterruptedException(toString());    }
    checkDeadLock();
    synchronized (this) {        while (!isDone()) {            incWaiters();            try {                wait();            } finally {                decWaiters();            }        }    }    return this;}

这里 while(!isDone()) 会进入循环，调用 sync() 后线程会被阻塞住。

总结

本篇也是写了好久，本文介绍了 ServerBootstrap，它是构建 Netty Server 的主要实现类，ServerBootstrap 里主要是对各种属性进行赋值，并创建 Channel 和 ChannelPipeline，最后绑定本地端口开始监听 IO 事件。在后续的文章里，我会继续与大家讨论 Netty 的 EventLoop，还请大家多多关注我的个人博客或公账号。

继续滑动看下一个