Netty 的 inbound 與 outbound, 以及 InboundHandler 的 channelInactive 與 OutboundHandler 的 close

先看一個例子.

有一個簡單 Server

public class SimpleServer {

    public static void main(String[] args) throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.SO_SNDBUF, 1024 * 1024)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new SimpleDuplex1());
                        ch.pipeline().addLast(new SimpleDuplex2());
                        ch.pipeline().addLast(new SimpleServerHandler());
                    }
                });
        b.bind(8090).sync().channel().closeFuture().sync();
    }
}

Handler 詳情如下

public class SimpleDuplex1 extends ChannelDuplexHandler {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("---- write 1 ----");
        super.write(ctx, msg, promise);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("---- read 1 ----");
        super.channelRead(ctx, msg);
    }
}

public class SimpleDuplex2 extends ChannelDuplexHandler {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("---- write 2 ----");
        super.write(ctx, msg, promise);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("---- read 2 ----");
        super.channelRead(ctx, msg);
    }
}

public class SimpleServerHandler extends ChannelDuplexHandler {

    @Override
    public void channelRead(ChannelHandlerContext ctx, final Object msg) throws Exception {
        ctx.channel().writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("OK".getBytes())).addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("----- INACTIVE -----");
        super.channelInactive(ctx);
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise future) throws Exception {
        System.out.println("----- CLOSE -----");
        super.close(ctx, future);
    }
}

啟動 Server 以后, 使用 telnet 發送數據查看執行結果

---- read 1 ----
---- read 2 ----
成功
---- write 2 ----
---- write 1 ----
----- CLOSE -----
----- INACTIVE -----

1. 先來看看執行順序, 可見, inbound 的順序是跟 add 順序一致的, 而 outbound 的順序是跟 add 順序相反的

以及, read 的 IO 觸發順序是 "socketChannel.read() -> 順序 handler -> TailContext.channelRead().releaseMsg"

而 write 的 IO 觸發順序是 "逆序 handler -> HeadContext.socketChannel.write()"

也就是說 read 是先觸發 socket 的 read IO 時間, 再進入 handler, 而如果我們最后一個 handler 未能完全處理消息, 調用了 super.channelRead, 則會進入 TailContext. 此時TailContext 會打出 debug 消息告訴你消息進入了最后一個 Handler 而未被處理. 因為一般來講都應該在自己的 handler 里把消息處理掉. 而不是讓他進入到默認 handler 里.

而對於 write 來說, 則是先進入自定義 handler, 最后在進入 HeadContext 觸發 IO 時間

2. 再來說說 close 與 channelInactive

前面說到了. Outbound 的順序是最后才執行到 HeadContext 來執行實際的 IO 操作, close 也是一樣, 當你調用 channle.close 的時候, 先會經過你的 handler . 最后調用 HeadContext.socketChannel.close(). 所以, 在我們的 Handler 中, 先會打印 "---- CLOSE ----" 然后再調用實際的 socketChannel.close. 最后, 當 close 成功時, 觸發 ChannelInactive 時間.

所以說 close 與 channelInactive 的關系是 close 是主動關閉 channel 的動作, 而 channelInactive 是關閉成功后收到通知的事件.