netty5笔记-总体流程分析4-NioServerSocketChannel

前面我们讲了server的启动过程，同时介绍了其中非常重要的几个概念,ChannelPipeline,ChannelHandlerContext等。接下来我们看看server启动起来以后是如何运转的。

先回忆下之前的几个重要的点，如果要监听某个端口，首先用ServerBootstrap引导启动，启动时创建一个ServerSocketChannel，并注册到bossGroup的EventLoop中，关注的事件为OP_ACCEPT。boss EventLoop开始运行，并不停的尝试从Selector中查看是否有准备好的连接。由于ServerSocketChannel关注的是OP_ACCEPT，因此每当有客户端连接到服务端的时候，boss EventLoop都可以select到一个SelectionKey，然后进入以下方法：

 

    //NioEventLoop
    private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
       final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            // 如果key无效了则关闭连接
            unsafe.close(unsafe.voidPromise());
            return;
        }

        try {
            int readyOps = k.readyOps();
            // 检查readOps，避免由jdk bug导致readOps为0而产生自旋(cpu 100%)
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                // 调用unsafe.read方法，这个也是我们今天分析的入口
                unsafe.read();
                if (!ch.isOpen()) {
                    // 连接已经关闭则直接返回，就不用处理后面的write事件了
                    return;
                }
            }
            。。。省略不会触发的几句代码。。。
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // 取消此key对 OP_CONNECT的关注，否则Selector.select(..)不会阻塞而直接返回，导致cpu 100%
                // 此bug见 https://github.com/netty/netty/issues/924
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);

                unsafe.finishConnect();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }

注意还有一种情况是processSelectedKey(SelectionKey k, NioTask<SelectableChannel> task)，这是扩展的情况，暂时无人实现，先不分析。

 

这里引出了本次分析的第一个入口点，ch.unsafe().read()。NioServerSocketChannel对应的unsafe实现为NioMessageUnsafe，我们来看看它的read方法做了哪些事情（去掉部分无关代码）：

 

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

        public void read() {
            final ChannelConfig config = config();
            // 下面有个循环操作，这里的值代表循环的最大次数，对于NioServerSocketChannel来说也就是单次最大接受的连接数，默认为16，
            // 可以在启动的时候通过初始化时调用引导类的setOption(ChannelOption.MAX_MESSAGES_PER_READ, 32)修改这个值。
           final int maxMessagesPerRead = config.getMaxMessagesPerRead();
            final ChannelPipeline pipeline = pipeline();
            boolean closed = false;
            Throwable exception = null;
            try {
                try {
                    for (;;) {
                        // 此处会调用到NioServerSocketChannel中的doReadMessages方法
                        int localRead = doReadMessages(readBuf);
                        // 读到的值为0没获取到连接（可能是已经关闭了），注意NioServerSocketChannel中的doReadMessages只会返回0,1，   
                        // -1在其他场景中出现，后面再分析                    
                        if (localRead == 0) {
                            break;
                        }
                        
                        // 每次读取成功readBuf中就会多一个连接，达到阈值则先跳出循环，剩下的数据下次循环时再取
                        if (readBuf.size() >= maxMessagesPerRead) {
                            break;
                        }
                    }
                } catch (Throwable t) {
                    exception = t;
                }
                setReadPending(false);
                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    // 对每个连接调用pipeline的fireChannelRead
                    pipeline.fireChannelRead(readBuf.get(i));
                }
                // 清理获取到的数据，下次继续使用该buf
                readBuf.clear();
                pipeline.fireChannelReadComplete();
                // 如果在接收连接时发生了错误，触发fireExceptionCaught事件
                if (exception != null) {
                    。。。。。。

                    pipeline.fireExceptionCaught(exception);
                }
                。。。。。。
           } finally {
                // 如果非autoRead则移除关注的事件
                if (!config.isAutoRead() && !isReadPending()) {
 removeReadOp();
 }
            }
        }

NioMessageUnsafe还是其他Channel的unsafe实现，这里只把NioServerSocketChannel相关的部分列了出来，逻辑很简单，循环的通过doReadMessages方法读取指定个数的连接到list中（如果其中一次没读取到也会终止循环），完成后对list中的连接依次调用pipeline.fireChannelRead方法。NioServerSocketChannel的doReadMessages比较简单：

 

 

    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = javaChannel().accept();

        try {
            if (ch != null) {
                // accept到连接则新建一个NioSocketChannel的封装类，并返回读取到的消息数1，否则返回0 
               buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
            logger.warn("Failed to create a new channel from an accepted socket.", t);

            try {
                // 创建失败则关闭改连接，这个连接指客户端连接            
                ch.close();
            } catch (Throwable t2) {
                logger.warn("Failed to close a socket.", t2);
            }
        }

        return 0;
    }

NioSocketChannel的创建过程我们后面再讲，先往后看，pipeline.fireChannelRead触发了什么方法。我们知道channelRead属于inbound时间，因此是从HeadContext->TailContext处理。HeadContext的fireChannelRead方法只是简单的往后传递事件，而TailContext则是对需要释放资源的对象进行释放，因此主要逻辑不用关注这两块。回到ServerBootstrap，在启动时默认会往pipeline中加入一个ServerBootstrapAcceptor（其他用户自定义加入的ChannelHandler如LoggingHandler我们暂不分析）。

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new ServerBootstrapAcceptor(
                        currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });

fireChannelRead主要的实现就是它了：

 

        // ServerBootstrapAcceptor
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            // child是对socket的一个包装、增强
            final Channel child = (Channel) msg;
            // 引导类启动时设置的childHandler加到child的pipeline中
            child.pipeline().addLast(childHandler);
            // 将childOptions中的配置设置到child的option中
            for (Entry<ChannelOption<?>, Object> e: childOptions) {
                try {
                    if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
                        logger.warn("Unknown channel option: " + e);
                    }
                } catch (Throwable t) {
                    logger.warn("Failed to set a channel option: " + child, t);
                }
            }
            // 将childAttrs中的属性设置到child的属性中
            for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
            }
            // 将连接注册到childGroup中（也就是我们常说的workGroup)，注册完成如果发现注册失败则关闭此链接
            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);
            }
        }

        private static void forceClose(Channel child, Throwable t) {
            child.unsafe().closeForcibly();
            logger.warn("Failed to register an accepted channel: " + child, t);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final ChannelConfig config = ctx.channel().config();
            if (config.isAutoRead()) {
                // 先设置为false，不再接收连接
               config.setAutoRead(false);
                // 1秒后设置回来
               ctx.channel().eventLoop().schedule(new Runnable() {
                    public void run() {
                       config.setAutoRead(true);
                    }
                }, 1, TimeUnit.SECONDS);
            }
            
            // 传递异常事件，这样用户可以自定义处理方法
           ctx.fireExceptionCaught(cause);
        }
    }

需要注意上面的exceptionCaught方法，从前面的NioMessageUnsafe.read()可以看到，其中的一个调用javaChannel.accept()并没有catch相应的异常，它可能会触发很多种异常，我们需要关注的比如IOException: Too many open files。对于这种情况，如果server继续接收连接则会触发更多的Too many open files，因此此处将autoRead设置为false，在设置为false后，NioMessageUnsafe的read方法会在finally中移除channel关注的事件：OP_CONNECT，这样server就不会再接收连接了。 同时也会添加一个计划任务，该任务1秒后执行，内容为重新将autoRead设为true。设置autoRead为true时会触发channel.read()方法，read方法接着触发pipeline.read()，最终触发HeadContext的read()，此方法调用了unsafe.beginRead()， beginRead将会再次将OP_CONNECT设置为关注的事件，此时可以继续接收客户端连接。

 

上面的channelRead方法通过childGroup.register(child)将客户端连接注册到了workGroup上，跟踪该方法进入到了SingleThreadEventLoop：

 

    public ChannelFuture register(final Channel channel, final ChannelPromise promise) {
        。。。

        channel.unsafe().register(this, promise);
        return promise;
    }

即调用了NioSocketChannel的unsafe.register, 此处的unsafe实现类为NioByteUnsafe，对应的register方法在其父类AbstractUnsafe中实现:

 

 

        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            if (eventLoop == null) {
                throw new NullPointerException("eventLoop");
            }
            if (promise == null) {
                throw new NullPointerException("promise");
            }
            // 一个连接不能多次注册，防止注册到其他EventLoop出现并发问题
           if (isRegistered()) {
                promise.setFailure(new IllegalStateException("registered to an event loop already"));
                return;
            }
            // 这里会判断是否时NioEventLoop，NioSocketChannel只能注册到NioEventLoop或者其子类
            if (!isCompatible(eventLoop)) {
                promise.setFailure(
                        new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
                return;
            }

            // It's necessary to reuse the wrapped eventloop object. Otherwise the user will end up with multiple
            // objects that do not share a common state.
            // PausableChannelEventLoop提供了一个开关isAcceptingNewTasks，isAcceptingNewTasks=false时不接收新任务
            if (AbstractChannel.this.eventLoop == null) {
                AbstractChannel.this.eventLoop = new PausableChannelEventLoop(eventLoop);
            } else {
                AbstractChannel.this.eventLoop.unwrapped = eventLoop;
            }

            // 线程安全的调用register0
            if (eventLoop.inEventLoop()) {
                register0(promise);
            } else {
                try {
                    eventLoop.execute(new OneTimeTask() {
                        @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);
                }
            }
        }

        private void register0(ChannelPromise promise) {
 try {
 // 将对应的promise设为不可取消
 if (!promise.setUncancellable() || !ensureOpen(promise)) {
 return;
 }
                // 调用doRegister进行注册
 boolean firstRegistration = neverRegistered;
 doRegister();
 neverRegistered = false;
                // register状态设置为true，表示已经注册到EventLoop中
 registered = true;
                // eventLoop的isAcceptingNewTasks开关打开
 eventLoop.acceptNewTasks();
 safeSetSuccess(promise);
                // 触发channelRegistered事件
 pipeline.fireChannelRegistered();
 // 第一次注册时触发fireChannelActive事件，防止deregister后再次register触发多次fireChannelActive调用
 if (firstRegistration && isActive()) {
                    // 这里和前面的ServerSocketChannel分析一样,最终会触发unsafe.beginRead()
 pipeline.fireChannelActive();
 }
 } catch (Throwable t) {
 // Close the channel directly to avoid FD leak.
 closeForcibly();
 closeFuture.setClosed();
 safeSetFailure(promise, t);
 }
 }

    // AbstractNioChannel
    protected void doRegister() throws Exception {
 boolean selected = false;
 for (;;) {
 try {
                // 将连接注册到selector，此时虽然有注册，但ops为0，即没有关注的事件
 selectionKey = javaChannel().register(((NioEventLoop) eventLoop().unwrap()).selector, 0, this);
 return;
 } catch (CancelledKeyException e) {
 if (!selected) {
                    // 强制Selector调用select now，防止取消的SelectionKey未真正取消（因为还没有调用到Select.select(..))
 ((NioEventLoop) eventLoop().unwrap()).selectNow();
 selected = true;
 } else {
 // We forced a select operation on the selector before but the SelectionKey is still cached
 // for whatever reason. JDK bug ?
 throw e;
 }
 }
 }
 }

以HttpSnoopServer为例，childHandler为HttpSnoopServerInitializer，初始化是NioSocketChannel的pipeline中只有tail,head及HttpSnoopServerInitializer的实例，tail和head中的channelRegisterd除了传递或者停止外没有其他用处，因此此处的channelRegisterd主要逻辑在HttpSnoopServerInitializer的channelRegistered方法：

 

    // 此方法在父类ChannelInitializer中
    public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ChannelPipeline pipeline = ctx.pipeline();
        boolean success = false;
        try {
            // initChannel主要是初始化channel对应的pipeline
           initChannel((C) ctx.channel());
            // 初始化完成后将自身移除
            pipeline.remove(this);
            // channelRegistered属于inbound事件，注册后调用一次该方法，这样所有用户添加的handler可以感知到此事件
            ctx.fireChannelRegistered();
            success = true;
        } catch (Throwable t) {
            logger.warn("Failed to initialize a channel. Closing: " + ctx.channel(), t);
        } finally {
            if (pipeline.context(this) != null) {
                pipeline.remove(this);
            }
            if (!success) {
                ctx.close();
            }
        }
    }
    
    // 此方法在HttpSnoopServerInitializer中
    public void initChannel(SocketChannel ch) {
 ChannelPipeline p = ch.pipeline();
 if (sslCtx != null) {
 p.addLast(sslCtx.newHandler(ch.alloc()));
 }
 p.addLast(new HttpRequestDecoder());
 // Uncomment the following line if you don't want to handle HttpChunks.
 //p.addLast(new HttpObjectAggregator(1048576));
 p.addLast(new HttpResponseEncoder());
 // Remove the following line if you don't want automatic content compression.
 //p.addLast(new HttpContentCompressor());
 p.addLast(new HttpSnoopServerHandler());
 }

而pipeline.fireChannelActive()最终会调用unsafe.beginRead()，我们看看NioSocketChannel的unsafe.beginRead()：

 

 

        public final void beginRead() {
            if (!isActive()) {
                return;
            }

            try {
                doBeginRead();
            } catch (final Exception e) {
                invokeLater(new OneTimeTask() {
                    public void run() {
                        pipeline.fireExceptionCaught(e);
                    }
                });
                close(voidPromise());
            }
        }

    // AbstractNioChannel
    protected void doBeginRead() throws Exception {
 if (inputShutdown) {
 return;
 }

 final SelectionKey selectionKey = this.selectionKey;
 if (!selectionKey.isValid()) {
 return;
 }

 readPending = true;
        // 注册NioSocketChannel关注的事件（NioSocketChannel初始关注的事件为OP_READ)
 final int interestOps = selectionKey.interestOps();
 if ((interestOps & readInterestOp) == 0) {
 selectionKey.interestOps(interestOps | readInterestOp);
 }
 }

到这里，一个NioSocketChannel已经可以读取数据了。 我们总结下这个过程：

 

1、bossGroup通过不停的selector.select或者selectNow获取到客户端连接（每个循环获取多个连接，批量处理），将socket包装为NioSocketChannel；

2、将新增的NioSocketChannel注册到workGroup中，该注册会触发以下操作；

2.1、首先将对应socket注册到workGroup对应的Selector中，此时关注的事件为空；

2.2、触发fireChannelRegistered事件， 该方法用用户自定义的handler及顺序来初始化channel的pipeline；

2.3、触发fireChannelActive事件，最终调用unsafe.beginRead()，beginRead方法设置NioSocketChannel关注的事件(OP_READ)。

过程比较简单，而且第2步的操作都是在workGroup中进行的，因此bossGroup的工作非常简单，所以效率很高。

下一篇我们将分析NioSocketChannel的整个生命周期，内容会比较多，花的时间也会比较长，敬请期待。。