Netty ChannelPipeline 数据管道

1. 前言

前面，我们也提到了 ChannelPipeline，它是管道或者说管理 Handler 的集合，很多同学很容易搞混 Channel、ChannelPipeline 和 ChannelHandler 之间关系。

本节内容我们需要理清并且掌握以下知识点：

1. Channel、ChannelPipeline、ChannelHandler 之间的关系；
2. 了解 ChannelPipeline 如何管理 ChannelHandler。

2. 三者之间的关系

Channel 是一个连接通道，客户端和服务端连接成功之后，会维持一个 Channel，可以通过 Channel 来发送数据。Channel 有且仅有一个 ChannelPipeline 与之相对应，ChannelPipeline 又维护着一个由多个 ChannelHandlerContext 组成的双向链表，ChannelHandlerContext 又关联着一个 ChannelHandler。

它们之间的关系，大概如下图所示：

3. ChannelPipeline 核心方法

ChannelPipeline 的最常用方法：

实例：

ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap
    .group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        protected void initChannel(NioSocketChannel ch) {
            ch.pipeline().addLast(new Handler1());
            ch.pipeline().addLast(new Handler2());
            ch.pipeline().addLast(new Handler3());
            ch.pipeline().addLast(new Handler4());
        }
    });

总结，ChannelPipeline 的用法比较固定，虽然方法很多，但是一般常用的就是 addLast。

4. ChannelPipeline 管理链表

实例：

//1.创建ChannelPipeline
ChannelPipeline pipeline = ch.pipeline();

//2.创建Handler
FirstHandler firstHandler = new FirstHandler();
SecondHandler secondHandler=new SecondHandler();
ThirdHandler thirdHandler=new ThirdHandler();
FourthHandler fourthHandler=new FourthHandler();

//3.操作
pipeline.addLast("handler1", firstHandler);
pipeline.addFirst("handler2", secondHandler);//在最开始添加
pipeline.addLast("handler3", thirdHandler);//在最后添加
pipeline.remove("handler3"); //根据名称删除
pipeline.remove(firstHandler);//根据对象删除
pipeline.replace("handler2", "handler4", fourthHandler);//替换

输出结果：

FourthHandler

5. 入站和出站执行顺序

在真实的项目开发当中，inboundHandler 和 outboundHandler 都是多个的，一般是一个业务处理对应一个 Handler。那么多个的情况下，Pipeline 的执行顺序又是怎么样的呢？

5.1 Inbound 不往下传递

实例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound1>>>>>>>>>");
    }
}

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
    }
}

执行结果：

inbound1>>>>>>>>>

思考：为什么不执行 InboundHandler2 呢？

原因：InboundHandler1 没有手工往下传递执行。

5.2 Inbound 流转顺序

实例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound1>>>>>>>>>");
        
        //往下传递
        super.channelRead(ctx, msg);
    }
}

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
    }
}

执行结果：

inbound1>>>>>>>>>
inbound2>>>>>>>>>

InboundHandler 之间可以通过 super.channelRead(ctx, msg); 往下传递。

5.3 Inbound 执行顺序

实例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //1.往下传递
        super.channelRead(ctx, msg);
        //2.打印信息
        System.out.println("inbound1>>>>>>>>>");
    }
}

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
    }
}

执行结果：

inbound2>>>>>>>>>
inbound1>>>>>>>>>

InboundHandler1 先往下传递，在执行自己的业务，那么 InboundHandler2 就会比 InboundHandler1 先执行。

总结：Inbound 是按顺序进行传递，但是逻辑的执行并非是按顺序执行，而是由 super.channelRead(ctx, msg); 去决定。

5.4 流转到 Outbound

InboundHandler 往 OutboundHandler 流转，需要手工调用 ctx.channel().writeAndFlush()，否则无法执行 OutboundHandler 的业务逻辑。

实例：

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
		//传递到OutboundHandler
        ctx.channel().writeAndFlush("hello world");
    }
}

5.5 Outbound 内部流转

跟 InboundHandler 一样，需要手工往下传递，否则无法流转到下一个 OutboundHandler。

实例：

public class OutboundHandler2 extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("outbound2>>>>>>>>>");
		//往下流转
        super.write(ctx, msg, promise);
    }
}

总结：OutboundHandler 是按逆向来流转，但是业务逻辑的执行顺序则是由 super.write(ctx, msg, promise); 决定。

5.6 ctx.writeAndFlush 和 ctx.channel ().writeAndFlush 的区别

很多同学很容易遇到以下问题，并且会想不通。

实例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());
ch.pipeline().addLast(new OutboundHandler1());
ch.pipeline().addLast(new OutboundHandler2());

InboundHandler2 流转代码

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
		//流转到OutboundHandler2
        ctx.channel().writeAndFlush("hello world");
    }
}

执行结果：

inbound1>>>>>>>>>
inbound2>>>>>>>>>
outbound2>>>>>>>>>
outbound1>>>>>>>>>

修改 InboundHandler2 流转代码

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
		//【注意，调整这里了】
        ctx.writeAndFlush("hello world");
    }
}

执行结果

inbound1>>>>>>>>>
inbound2>>>>>>>>>

思考：为什么这里使用 ctx.writeAndFlush 就流程不下去了呢？

ctx.writeAndFlush(); 最终源码

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

通过源码，我们发现它是从当前 InboundHandler 开始往前执行。

ctx.channel().writeAndFlush(); 最终源码

public final ChannelFuture writeAndFlush(Object msg) {
    return this.tail.writeAndFlush(msg);
}

通过源码，我们发现它是从链表的最后一个节点开始往前面执行。

总结，如果是 OutboundHandler 放在 InboundHandler 之后，使用不同的 writeAndFlush 则得到的结果不一样。

5.7 规律总结

Inbound 的顺序

1. 流转顺序： 多个 Inbound 不会自动往下流转，需要手工调用 ctx.fireChannelRead(msg); 才能流转到下一个；
2. 执行顺序： 业务逻辑的执行顺序，则根据 ctx.fireChannelRead(msg); 和逻辑的先后顺序所决定；
3. Inbound 往 Outbound 流转，则需要手工 ctx.channel().writeAndFlush()。

Outbound 的顺序

1. 流转顺序： 多个 Outbound 不会自动往下流转，需要手工调用 ctx.write(msg, promise); 才能流转到下一个；
2. 执行顺序： 业务逻辑的执行顺序，则根据 ctx.write(msg, promise); 和逻辑的先后顺序所决定。

6. 小结

本文主要讲解的知识点

1. Channel、ChannelPipeline、ChannelHandlerContext、ChannelHandler 之间的关系；
2. ChannelPipeline 的核心方法，以及它是如何管理 Handler 的，主要通过 addLast () 去组装 Handler；
3. 入站和出站的流转顺序和业务逻辑的执行顺序。