写在前面

服务既然有启动，那么根据爱因斯坦的相对论，就肯定有停止，所以本文就来看下netty停止服务相关源码分析。

1：准备

为了调试我们需要稍微改造下netty源码中example模块的echoserver类，修改为如下:

public final class EchoServerForStopServerDebug {

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

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

        // Configure the server.,这里的线程数就设置为1了，正常如果只监听一个端口号只需要一个
        // ，而且源码也是只会选择一个EventLoop
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            if (sslCtx != null) {
                                p.addLast(sslCtx.newHandler(ch.alloc()));
                            }
                            //p.addLast(new LoggingHandler(LogLevel.INFO));
                            p.addLast(serverHandler);
                        }
                    });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
//            f.channel().closeFuture().sync(); // 注释不然服务会一直阻塞
            Thread.sleep(30000); // 必须休眠才会处理客户端的连接，当阻塞在bossGroup.shutdownGracefully();断点时不会
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

正戏

先在如下位置打断点：

接着启动echoclient类，稍微等一会，等待进入到上图的断点,进入断点后执行到如下代码：

// io.netty.util.concurrent.MultithreadEventExecutorGroup#shutdownGracefully
public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
    for (EventExecutor l: children) { // children的个数就是new NioEventLoopGroup时的线程数，默认不指定是核数*2
        l.shutdownGracefully(quietPeriod, timeout, unit);
    }
    return terminationFuture();
}

方法shutdownGracefully：

// io.netty.util.concurrent.SingleThreadEventExecutor#shutdownGracefully
public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
    // ...
    // 通过状态判断是否正在关闭中，避免重复调用（幂等处理）
    if (isShuttingDown()) {
        return terminationFuture();
    }
    // 是否是evenloop线程调用，因为一般是在启动函数的finally中调用，所以一般是false
    boolean inEventLoop = inEventLoop();
    boolean wakeup;
    int oldState;
    for (;;) {
        if (isShuttingDown()) { // for循环外已经判断了一次，这里再次判断的原因是for循环后边的代码会改变状态，所以，这里的判断是for死循环的终止条件
            return terminationFuture();
        }
        int newState;
        wakeup = true;
        oldState = state; // 一般是 private static final int ST_STARTED = 2;
        if (inEventLoop) {
            newState = ST_SHUTTING_DOWN;
        } else {
            switch (oldState) {
                case ST_NOT_STARTED:
                case ST_STARTED: // ST_STARTED -> ST_SHUTTING_DOWN
                    newState = ST_SHUTTING_DOWN;
                    break;
                default:
                    newState = oldState;
                    wakeup = false;
            }
        }
        if (STATE_UPDATER.compareAndSet(this, oldState, newState)) { // 真正的修改state的状态值
            break;
        }
    }
    // 记录静默时间剩余量，和优雅关闭时间剩余量，因为刚关闭，自然指定的时间就是了，只不过做一个时间转换
    gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
    gracefulShutdownTimeout = unit.toNanos(timeout);

    // ...
}

上述代码修改了服务的状态值state，并且记录了一些用于优雅关闭的时间等信息，其中服务状态值的改变比较重要，这里将该状态值修改为ST_SHUTTING_DOWN，在NioEventLoop的run死循环中就会获取到这个状态的改变，开始停止自己：

// io.netty.channel.nio.NioEventLoop#run
protected void run() {
    int selectCnt = 0;
    for (;;) {
        try {
            // ...
        } catch (CancelledKeyException e) {
            // ...
        } catch (Error e) {
            // ...
        } catch (Throwable t) {
            // ...
        } finally {
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) { // 调用了shutdownGracefully
                    closeAll(); // 关闭注册在当前selector上所有的channel
                    if (confirmShutdown()) { // 优雅关闭，给正在执行的任务预留一点点时间，但不保证绝对执行完，因为有最大退出时间做控制
                        return; // return，run的死循环就结束了，对应的线程也就结束了，当所有的线程都结束了，jvm也就自动退出了
                    }
                }
            } catch (Error e) {
                throw (Error) e;
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
}

方法isShuttingDown()就会检测到state状态值的变更，开始停止eventloop的逻辑，当所有的event loop都停止了，服务也就停止了。confirmShutdown()是优雅关闭的方法，并不会无脑直接停止，而是优雅的稍微等一会，closeAll是关闭通道，关闭selector等，如下：

// io.netty.channel.nio.NioEventLoop#closeAll
private void closeAll() {
    selectAgain(); // 去除cancel的key，只保留有效的key，什么时候就会cancel呢，比如客户端主动关闭了连接
    Set<SelectionKey> keys = selector.keys();
    Collection<AbstractNioChannel> channels = new ArrayList<AbstractNioChannel>(keys.size());
    for (SelectionKey k: keys) {
        Object a = k.attachment(); // 这个a就是用于给客户端写数据的socket channel了
        if (a instanceof AbstractNioChannel) {
            channels.add((AbstractNioChannel) a);
        } else {
            k.cancel();
            @SuppressWarnings("unchecked")
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            invokeChannelUnregistered(task, k, null);
        }
    }
    // 关闭channel 详细可参考：https://dongyunqi.blog.csdn.net/article/details/143687372
    for (AbstractNioChannel ch: channels) {
        ch.unsafe().close(ch.unsafe().voidPromise());
    }
}

写在后面

参考文章列表