chenssy
致力打造Java精品博客站点

【死磕Netty】—–服务端启动过程分析

原文出处http://cmsblogs.com/ 『chenssy』
转载请注明原创出处,谢谢!

上篇博客(【死磕Netty】----Netty的核心组件及其设计),了解了 Netty 的核心组件及其设计,但是这些都是零散的,不成体系。那么 Netty 是如何利用这些组件构建成一个高性能的异步通信框架。通过这篇博客可以初步了解。

下面先来一段 Netty 服务端的代码:

public class NettyServer {

    public void bind(int port){
        // 创建EventLoopGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup();        //创建BOSS线程组 用于服务端接受客户端的连接
        EventLoopGroup workerGroup = new NioEventLoopGroup();      //创建WORK线程组 用于进行SocketChannel的网络读写

        try {
            // 创建ServerBootStrap实例
            // ServerBootstrap 用于启动NIO服务端的辅助启动类,目的是降低服务端的开发复杂度
            ServerBootstrap b = new ServerBootstrap();
            // 绑定Reactor线程池
            b.group(bossGroup, workerGroup)
                    // 设置并绑定服务端Channel
                    // 指定所使用的NIO传输的Channel
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .handler(new LoggingServerHandler())
                    .childHandler(new ChannelInitializer(){

                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            //do something
                        }
                    });

            // 绑定端口,同步等待成功
            ChannelFuture future = b.bind(port).sync();
            // 等待服务端监听端口关闭
            future.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 优雅地关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    private class LoggingServerHandler extends ChannelInboundHandlerAdapter{
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("loggin-channelActive");
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            System.out.println("loggin-channelRegistered");
        }

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            System.out.println("loggin-handlerAdded");
        }
    }

    public static void main(String[] args){
            new NettyServer().bind(8899);
    }
}

上面代码为 Netty 服务器端的完整代码,在整个服务端代码中会涉及如下几个核心类。

ServerBootstrap

ServerBootstrap 为 Netty 服务端的启动辅助类,它提供了一系列的方法用于设置服务端启动相关的参数。

Channel

Channel 为 Netty 网络操作抽象类,它定义了一组功能,其提供的 API 大大降低了直接使用 Socket 类的复杂性。当然它也不仅仅只是包括了网络 IO 操作的基本功能,还包括一些与 Netty 框架相关的功能,包括获取该 Channel 的 EventLoop 等等。

EventLoopGroup

EventLoopGroup 为 Netty 的 Reactor 线程池,它实际上就是 EventLoop 的容器,而 EventLoop 为 Netty 的核心抽象类,它的主要职责是处理所有注册到本线程多路复用器 Selector 上的 Channel。

ChannelHandler

ChannelHandler 作为 Netty 的主要组件,它主要负责 I/O 事件或者 I/O 操作进行拦截和处理,它可以选择性地拦截和处理自己感觉兴趣的事件,也可以透传和终止事件的传递。

ChannelPipeline

ChannelPipeline 是 ChannelHandler 链的容器,它负责 ChannelHandler 的管理和事件拦截与调度。每当新建一个 Channel 都会分配一个新的 ChannelPepeline,同时这种关联是永久性的。

以上是简要介绍,详细介绍请参考(【死磕Netty】-----Netty的核心组件及其设计)

服务端创建流程

Netty 服务端创建的时序图,如下(摘自《Netty权威指南(第二版)》)

Netty 服务端创建的时序图

主要步骤为:

  1. 创建 ServerBootstrap 实例
  2. 设置并绑定 Reactor 线程池
  3. 设置并绑定服务端 Channel
  4. 创建并初始化 ChannelPipeline
  5. 添加并设置 ChannelHandler
  6. 绑定并启动监听端口

服务端源码分析

1、创建两个EventLoopGroup

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

bossGroup 为 BOSS 线程组,用于服务端接受客户端的连接, workerGroup 为 worker 线程组,用于进行 SocketChannel 的网络读写。当然也可以创建一个并共享。

2、创建ServerBootstrap实例

ServerBootstrap b = new ServerBootstrap();

ServerBootStrap为Netty服务端的启动引导类,用于帮助用户快速配置、启动服务端服务。提供的方法如下:

方法名称 方法描述
group 设置 ServerBootstrap 要用的 EventLoopGroup
channel 设置将要被实例化的 ServerChannel 类
option 实例化的 ServerChannel 的配置项
childHandler 设置并添加 ChannelHandler
bind 绑定 ServerChannel

ServerBootStrap底层采用装饰者模式。

关于 ServerBootStrap 我们后续做详细分析。

3、设置并绑定Reactor线程池

调用 group() 方法,为 ServerBootstrap 实例设置并绑定 Reactor 线程池。

b.group(bossGroup, workerGroup)

EventLoopGroup 为 Netty 线程池,它实际上就是 EventLoop 的数组容器。EventLoop 的职责是处理所有注册到本线程多路复用器 Selector 上的 Channel,Selector 的轮询操作由绑定的 EventLoop 线程 run 方法驱动,在一个循环体内循环执行。通俗点讲就是一个死循环,不断的检测 I/O 事件、处理 I/O 事件。

这里设置了两个group,这个其实有点儿像我们工作一样。需要两类型的工人,一个老板(bossGroup),一个工人(workerGroup),老板负责从外面接活,工人则负责死命干活(尼玛,和我上家公司一模一样)。所以这里 bossGroup 的作用就是不断地接收新的连接,接收之后就丢给 workerGroup 来处理,workerGroup 负责干活就行(负责客户端连接的 IO 操作)。

源码如下:

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

其中父 EventLoopGroup 传递到父类的构造函数中:

    public B group(EventLoopGroup group) {
        if (group == null) {
            throw new NullPointerException("group");
        }
        if (this.group != null) {
            throw new IllegalStateException("group set already");
        }
        this.group = group;
        return (B) this;
    }

4、设置并绑定服务端Channel
绑定线程池后,则需要设置 channel 类型,服务端用的是 NioServerSocketChannel 。

.channel(NioServerSocketChannel.class)

调用 ServerBootstrap.channel 方法用于设置服务端使用的 Channel,传递一个 NioServerSocketChannel Class对象,Netty通过工厂类,利用反射创建NioServerSocketChannel 对象,如下:

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

channelFactory() 用于设置 Channel 工厂的:

    public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) {
        return channelFactory((ChannelFactory<C>) channelFactory);
    }

    public B channelFactory(ChannelFactory<? extends C> channelFactory) {
        if (channelFactory == null) {
            throw new NullPointerException("channelFactory");
        }
        if (this.channelFactory != null) {
            throw new IllegalStateException("channelFactory set already");
        }

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

这里传递的是 ReflectiveChannelFactory,其源代码如下:

public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {

    private final Class<? extends T> clazz;

    public ReflectiveChannelFactory(Class<? extends T> clazz) {
        if (clazz == null) {
            throw new NullPointerException("clazz");
        }
        this.clazz = clazz;
    }
    //需要创建 channel 的时候,该方法将被调用
    @Override
    public T newChannel() {
        try {
            // 反射创建对应 channel
            return clazz.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + clazz, t);
        }
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(clazz) + ".class";
    }
}

确定服务端的 Channel(NioServerSocketChannel)后,调用 option()方法设置 Channel 参数,作为服务端,主要是设置TCP的backlog参数,如下:

.option(ChannelOption.SO_BACKLOG, 1024)

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

    private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();

五、添加并设置ChannelHandler

设置完 Channel 参数后,用户可以为启动辅助类和其父类分别指定 Handler。

 .handler(new LoggingServerHandler())
.childHandler(new ChannelInitializer(){
    //省略代码
})

这两个 Handler 不一样,前者(handler())设置的 Handler 是服务端 NioServerSocketChannel的,后者(childHandler())设置的 Handler 是属于每一个新建的 NioSocketChannel 的。跟踪源代码会发现两种所处的类不一样,handler 位于 AbstractBootstrap 中,childHandler 位于 ServerBootstrap 中,如下:

    // AbstractBootstrap
    public B handler(ChannelHandler handler) {
        if (handler == null) {
            throw new NullPointerException("handler");
        }
        this.handler = handler;
        return (B) this;
    }

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

ServerBootstrap 中的 Handler 是 NioServerSocketChannel 使用的,所有连接该监听端口的客户端都会执行它,父类 AbstractBootstrap 中的 Handler 是一个工厂类,它为每一个新接入的客户端都创建一个新的 Handler。如下图(《Netty权威指南(第二版)》):

这里写图片描述六、绑定端口,启动服务

服务端最后一步,绑定端口并启动服务,如下:

ChannelFuture future = b.bind(port).sync();

调用 ServerBootstrap 的 bind() 方法进行端口绑定:

    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }

    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }    

首先调用 validate() 方法进行参数校验,然后调用 doBind() 方法:

    private ChannelFuture doBind(final SocketAddress localAddress) {
        // 初始化并注册一个Channel
        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绑定
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            // Registration future is almost always fulfilled already, but just in case it's not.
            final AbstractBootstrap.PendingRegistrationPromise promise = new AbstractBootstrap.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;
        }
    }

该方法涉及内容较多,我们分解来看,如下:

  1. 首先通过 initAndRegister() 得到一个 ChannelFuture 对象 regFuture;
  2. 根据得到的 regFuture 对象判断该对象是否抛出异常 (regFuture.cause()),如果是,直接返回;
  3. 根据 regFuture.isDone()判断 initAndRegister()是否执行完毕,如果执行完成,则调用 doBind0
  4. initAndRegister() 没有执行完毕,则向 regFuture 对象添加一个 ChannelFutureListener 监听,当 initAndRegister() 执行完毕后会调用 operationComplete(),在 operationComplete() 中依然会判断 ChannelFuture 是否抛出异常,如果没有则调用 doBind0进行绑定。

按照上面的步骤我们一步一步来剖析 doBind() 方法。

initAndRegister()

执行 initAndRegister() 会得到一个 ChannelFuture 对象 regFuture,代码如下:

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            // 新建一个Channel
            channel = channelFactory.newChannel();
            // 初始化Channel
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                channel.unsafe().closeForcibly();
            }
            return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
        }

        // /向EventLoopGroup中注册一个channel
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }
        return regFuture;
    }

首先调用 newChannel() 新建一个Channel,这里是NioServerSocketChannel,还记前面 4、设置并绑定服务端Channel(.channel(NioServerSocketChannel.class)中 设置的Channel工厂类么?在这里派上用处了。在上面提到了通过反射的机制我们可以得到一个 NioServerSocketChannel 类的实例。那么 NioServerSocketChannel 到底是一个什么东西呢?如下图:

这里写图片描述

上图是 NioServerSocketChannel 的继承体系结构图, NioServerSocketChannel 在构造函数中会依靠父类来完成一项一项的初始化工作。先看 NioServerSocketChannel 构造函数。

    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }

newSocket() 方法较为简单,它是利用 SelectorProvider.openServerSocketChannel(),产生一个 ServerSocketChannel 对象。

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

该构造函数首先是调用父类的构造方法,然后设置 config属性。父类构造方法如下:

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

    // AbstractNioChannel
    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);
        }
    }

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

通过 super() ,一层一层往上,直到 AbstractChannel。我们从最上层解析。

  • AbstractChannel 设置了 unsafe (unsafe = newUnsafe())和 pipeline(pipeline = newChannelPipeline());
  • AbstractNioChannel 将当前 ServerSocketChannel 设置成了非阻塞(ch.configureBlocking(false);),同时设置SelectionKey.OP_ACCEPT事件(this.readInterestOp = readInterestOp; readInterestOp 值由 NioServerSocketChannel 中传递);
  • NioServerSocketChannel 设置 config属性(config = new NioServerSocketChannelConfig(this, javaChannel().socket()))。

所以 channel = channelFactory.newChannel() 通过反射机制产生了 NioServerSocketChannel 类实例。同时该实例设置了NioMessageUnsafe、DefaultChannelPipeline、非阻塞、SelectionKey.OP_ACCEPT事件 和 NioServerSocketChannelConfig 属性。

看完了 channelFactory.newChannel();,我们再看 init()

    void init(Channel channel) throws Exception {
         // 设置配置的option参数
        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());
            }
        }

        // 获取绑定的pipeline
        ChannelPipeline p = channel.pipeline();

        // 准备child用到的4个part
        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()));
        }

        // 为NioServerSocketChannel的pipeline添加一个初始化Handler,
        // 当NioServerSocketChannel在EventLoop注册成功时,该handler的init方法将被调用
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                //如果用户配置过Handler
                if (handler != null) {
                    pipeline.addLast(handler);
                }

                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        // 为NioServerSocketChannel的pipeline添加ServerBootstrapAcceptor处理器
                        // 该Handler主要用来将新创建的NioSocketChannel注册到EventLoopGroup中
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

其实整个过程可以分为三个步骤:

  1. 设置 Channel 的 option 和 attr;
  2. 获取绑定的 pipeline,然后为 NioServerSocketChanne l绑定的 pipeline 添加 Handler;
  3. 将用于服务端注册的 Handler ServerBootstrapAcceptor 添加到 ChannelPipeline 中。ServerBootstrapAcceptor 为一个接入器,专门接受新请求,把新的请求扔给某个事件循环器。

至此初始化部分已经结束,我们再看注册部分,

        // /向EventLoopGroup中注册一个channel
        ChannelFuture regFuture = config().group().register(channel);

注册方法的调用位于 initAndRegister() 方法中。注意这里的 group() 返回的是前面的 boss NioEvenLoopGroup,它继承 MultithreadEventLoopGroup,调用的 register(),也是 MultithreadEventLoopGroup 中的。如下:

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

调用 next() 方法从 EventLoopGroup 中获取下一个 EventLoop,调用 register() 方法注册:

    public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }

将Channel和EventLoop封装成一个DefaultChannelPromise对象,然后调用register()方法。DefaultChannelPromis为ChannelPromise的默认实现,而ChannelPromisee继承Future,具备异步执行结构,绑定Channel,所以又具备了监听的能力,故而ChannelPromis是Netty异步执行的核心接口。

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

首先获取 channel 的 unsafe 对象,该 unsafe 对象就是在之前设置过得。然后调用 register() 方法,如下:

        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;

            // 必须要保证注册是由该EventLoop发起的
            if (eventLoop.inEventLoop()) {
                register0(promise);        // 注册
            } else {
                // 如果不是单独封装成一个task异步执行
                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);
                }
            }
        }

过程如下:

  1. 首先通过isRegistered() 判断该 Channel 是否已经注册到 EventLoop 中;
  2. 通过 eventLoop.inEventLoop() 来判断当前线程是否为该 EventLoop 自身发起的,如果是,则调用 register0() 直接注册;
  3. 如果不是,说明该 EventLoop 中的线程此时没有执行权,则需要新建一个线程,单独封装一个 Task,而该 Task 的主要任务则是执行 register0()

无论当前 EventLoop 的线程是否拥有执行权,最终都会要执行 register0(),如下:

        private void register0(ChannelPromise promise) {
            try {
                // 确保 Channel 处于 open
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                boolean firstRegistration = neverRegistered;

                // 真正的注册动作
                doRegister();

                neverRegistered = false;
                registered = true;        

                pipeline.invokeHandlerAddedIfNeeded();    
                safeSetSuccess(promise);        //设置注册结果为成功

                pipeline.fireChannelRegistered();

                if (isActive()) { 
                    //如果是首次注册,发起 pipeline 的 fireChannelActive
                    if (firstRegistration) {
                        pipeline.fireChannelActive();
                    } else if (config().isAutoRead()) {
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }

如果 Channel 处于 open 状态,则调用 doRegister() 方法完成注册,然后将注册结果设置为成功。最后判断如果是首次注册且处于激活状态,则发起 pipeline 的 fireChannelActive()

    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                // 注册到NIOEventLoop的Selector上
                selectionKey = javaChannel().register(eventLoop().selector, 0, this);
                return;
            } catch (CancelledKeyException e) {
                if (!selected) {
                    eventLoop().selectNow();
                    selected = true;
                } else {
                    throw e;
                }
            }
        }
    }

这里注册时 ops 设置的是 0,也就是说 ServerSocketChannel 仅仅只是表示了注册成功,还不能监听任何网络操作,这样做的目的是(摘自《Netty权威指南(第二版)》):

  1. 注册方式是多态的,它既可以被 NIOServerSocketChannel 用来监听客户端的连接接入,也可以注册 SocketChannel 用来监听网络读或者写操作。
  2. 通过 SelectionKey.interestOps(int ops) 方法可以方便地修改监听操作位。所以,此处注册需要获取 SelectionKey 并给 AbstractNIOChannel 的成员变量 selectionKey 赋值。

由于这里 ops 设置为 0,所以还不能监听读写事件。调用 doRegister()后,然后调用pipeline.invokeHandlerAddedIfNeeded();,这个时候控制台会出现 loggin-handlerAdded,内部如何调用,我们在剖析 pipeline 时再做详细分析。然后将注册结果设置为成功(safeSetSuccess(promise))。调用 pipeline.fireChannelRegistered(); 这个时候控制台会打印 loggin-channelRegistered。这里简单分析下该方法。

    public final ChannelPipeline fireChannelRegistered() {
        AbstractChannelHandlerContext.invokeChannelRegistered(head);
        return this;
    }

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

pipeline 维护着 handle 链表,事件会在 NioServerSocketChannel 的 pipeline 中传播。最终都会调用 next.invokeChannelRegistered(),如下:

    private void invokeChannelRegistered() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRegistered(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRegistered();
        }
    }

invokeChannelRegistered() 会调用我们在前面设置的 handler (还记得签名的 handler(new LoggingServerHandler() )么)的 channelRegistered(),这个时候控制台应该会打印 loggin-channelRegistered

到这里initAndRegister() (final ChannelFuture regFuture = initAndRegister();)就分析完毕了,该方法主要做如下三件事:

  1. 通过反射产生了一个 NioServerSocketChannle 对象;
  2. 调用 init(channel)完成初始化工作;
  3. 将NioServerSocketChannel进行了注册。

initAndRegister()篇幅较长,分析完毕了,我们再返回到doBind(final SocketAddress localAddress)。在 doBind(final SocketAddress localAddress) 中如果 initAndRegister()执行完成,则 regFuture.isDone() 则为 true,执行 doBind0()。如果没有执行完成,则会注册一个监听 ChannelFutureListener,当 initAndRegister() 完成后,会调用该监听的 operationComplete()方法,最终目的还是执行 doBind0()。故而我们下面分析 doBind0()到底做了些什么。源码如下:

    private static void doBind0(
            final ChannelFuture regFuture, final Channel channel,
            final SocketAddress localAddress, final ChannelPromise promise) {

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

doBind0() 较为简单,首先new 一个线程 task,然后将该任务提交到 NioEventLoop 中进行处理,我们先看 execute()

  public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }

        boolean inEventLoop = inEventLoop();
        if (inEventLoop) {
            addTask(task);
        } else {
            startThread();
            addTask(task);
            if (isShutdown() && removeTask(task)) {
                reject();
            }
        }

        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }

调用 inEventLoop() 判断当前线程是否为该 NioEventLoop 所关联的线程,如果是,则调用 addTask() 将任务 task 添加到队列中,如果不是,则先启动线程,在调用 addTask() 将任务 task 添加到队列中。addTask() 如下:

    protected void addTask(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (!offerTask(task)) {
            reject(task);
        }
    }

offerTask()添加到队列中:

    final boolean offerTask(Runnable task) {
        if (isShutdown()) {
            reject();
        }
        return taskQueue.offer(task);
    }

task 添加到任务队列 taskQueue成功后,执行任务会调用如下方法:

 channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

channel 首先调用 bind() 完成 channel 与端口的绑定,如下:

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

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

tail 在 DefaultChannelPipeline 中定义:final AbstractChannelHandlerContext tail; 有 tail 就会有 head ,在 DefaultChannelPipeline 中维护这一个 AbstractChannelHandlerContext 节点的双向链表,该链表是实现 Pipeline 机制的关键,更多详情会在 ChannelPipeline 中做详细说明。bind() 最终会调用 DefaultChannelPipeline 的 bind() 方法。如下:

    public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        if (!validatePromise(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;
    }

首先对 localAddress 、 promise 进行校验,符合规范则调用 findContextOutbound() ,该方法用于在 pipeline 中获取 AbstractChannelHandlerContext 双向链表中的一个节点,如下:

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

从该方法可以看出,所获取的节点是从 tail 开始遍历,获取第一个节点属性 outbound 为 true 的节点。其实该节点是 AbstractChannelHandlerContext 双向链表的 head 节点。获取该节点后,调用 invokeBind(),如下:

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

handler() 返回的是 HeadContext 对象,然后调用其bind(),如下:

        public void bind(
                ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
                throws Exception {
            unsafe.bind(localAddress, promise);
        }

unsafe 定义在 HeadContext 中,在构造函数中初始化(unsafe = pipeline.channel().unsafe();),调用 bind() 如下:

        public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
            assertEventLoop();

            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }

            if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
                localAddress instanceof InetSocketAddress &&
                !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
                !PlatformDependent.isWindows() && !PlatformDependent.isRoot()) {

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

内部调用 doBind() ,该方法为绑定中最核心的方法,位于 NioServerSocketChannel 中,如下:

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

javaChannel()返回的是 NioServerSocketChannel 实例初始化时所产生的 Java NIO ServerSocketChannel 实例(ServerSocketChannelImple实例),然后调用其 bind(),如下:

    public ServerSocketChannel bind(SocketAddress var1, int var2) throws IOException {
        Object var3 = this.lock;
        synchronized(this.lock) {
            if(!this.isOpen()) {
                throw new ClosedChannelException();
            } else if(this.isBound()) {
                throw new AlreadyBoundException();
            } else {
                InetSocketAddress var4 = var1 == null?new InetSocketAddress(0):Net.checkAddress(var1);
                SecurityManager var5 = System.getSecurityManager();
                if(var5 != null) {
                    var5.checkListen(var4.getPort());
                }

                NetHooks.beforeTcpBind(this.fd, var4.getAddress(), var4.getPort());
                Net.bind(this.fd, var4.getAddress(), var4.getPort());
                Net.listen(this.fd, var2 < 1?50:var2);
                Object var6 = this.stateLock;
                synchronized(this.stateLock) {
                    this.localAddress = Net.localAddress(this.fd);
                }

                return this;
            }
        }
    }

该方法属于 Java NIO 层次的,该方法涉及到服务端端口的绑定,端口的监听,这些内容在后续的 Channel 时做详细介绍。

到这里就真正完成了服务端端口的绑定。

这篇博客比较长,大体上从源码层次稍微解读了 Netty 服务端的启动过程,当中涉及到 Netty 的各个核心组件,只能笼统来描述服务端的启动过程,具体的细节部分还需要后续做详细分析,而且其中有多个点还是懵懵懂懂,相信在后面对 Netty 的分析过程会一一解答。

谢谢阅读,祝好!!!

参考资料

如未加特殊说明,此网站文章均为原创,转载必须注明出处。cmsblogs-chenssy » 【死磕Netty】—–服务端启动过程分析

分享到:更多 ()

Chenssy's Blog | 致力打造个人精品博客

联系作者读书系统