Netty ChannelPipeline组件作用

ChannelPipeline是ChannelHandler的容器，类似于Servlet的Filter过滤器，它负责处理和拦截 inbound（入站） 或者 outbound（出站） 的事件和操作；(也可以这样理解：ChannelPipeline 是 保存 ChannelHandler 的 List，拦截穿过 Channel 的输入输出 event, ChannelPipeline 实现了拦截器的一种高级形式，使得用户可以对事件的处理以及ChannelHanler之间交互获得完全的控制权)；

每一个新创建的 Channel（也可以说是每创建一个新的连接） 都将会被分配一个新的 ChannelPipeline，这项关联是永久性的； Channel 既不能附加另外一个 ChannelPipeline，也不能分离其当前的；在 Netty 组件的生命周期中，这是一项固定的操作，不需要开发人员的任何干预；

//把一个业务处理类（handler）添加到链中的第一个位置
ChannelPipeline addFirst(ChannelHandler... handlers);

//把一个业务处理类（handler）添加到链中的最后一个位置
ChannelPipeline addLast(ChannelHandler... handlers);

　　

事件将会被 ChannelInboundHandler 或者 ChannelOutboundHandler，处理；随后， 通过调用 ChannelHandlerContext 实现，它将被转发给同一超类型的下一个ChannelHandler；

ChannelHandlerContext使得ChannelHandler能够和它的ChannelPipeline以及其他的ChannelHandler 交互；保存Channel相关的所有上下文信息，同时关联一个ChannelHandler对象；即ChannelHandlerContext中包含一个具体的事件处理器 ChannelHandler，同时ChannelHandlerContext中也绑定了对应的pipeline和channel的信息，方便对 ChannelHandler进行调用；

  

通过 ChannelHandlerContext 触发的操作的事件流，ChannelHandler可以通知其所属的ChannelPipeline 中 的下一 个ChannelHandler，甚至可以动态修改它所属的ChannelPipeline；

虽然被调用的 Channel 或 ChannelPipeline 上的 write()方法将一直传播事件通过整个 ChannelPipeline，但是在 ChannelHandler 的级别上，事件从一个 ChannelHandler到下一个 ChannelHandler 的移动是由 ChannelHandlerContext 上的调用完成的；类似下面这张图；

对于客户端和服务端，入站和出站是相对的；以客户端应用程序为例，如果事件的传播方向是从客户端到服务端的（发送请求到服务端），那么对于客户端为出站的（即客户端发送给服务端的数据会通过pipeline中的一系列ChannelOutboundHandler，并被这些Handler处理），对于服务端是入站的；

Channel、ChannelPipeline、ChannelHandlerContext 都可以调用此方法，前两者的事件传播会经过整个ChannelPipeline，而ChannelHandlerContext就只会在后续的Handler里面传播；

ChannelInboundHandler之间的传递，主要通过调用ChannelHandlerContext里面的FireXXX()方法来实现下个Handler的调用；

关于有多个入站出站ChannelHandler的执行顺序，测试如下；

• 服务端

　　服务端启动类添加Handler

serverBootstrap.group(bossGroup, workGroup)
    .channel(NioServerSocketChannel.class)
    .option(ChannelOption.SO_BACKLOG,1024)
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new InboundHandler1());
            ch.pipeline().addLast(new InboundHandler2());
            ch.pipeline().addLast(new OutboundHandler1());
            ch.pipeline().addLast(new OutboundHandler2());
        }
    });

　　OutboundHandler1

public class OutboundHandler1 extends ChannelOutboundHandlerAdapter {
    private final static Logger log = LoggerFactory.getLogger(OutboundHandler1.class);

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

        ByteBuf data = (ByteBuf) msg;
        log.info("|OutboundHandler1 write : "+data.toString(CharsetUtil.UTF_8));
        ctx.write(Unpooled.copiedBuffer("OutboundHandler1 "+data.toString(CharsetUtil.UTF_8), CharsetUtil.UTF_8));
        ctx.flush();
    }
}

　　OutboundHandler2

public class OutboundHandler2 extends ChannelOutboundHandlerAdapter {
    private final static Logger log = LoggerFactory.getLogger(OutboundHandler2.class);

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

        ByteBuf data = (ByteBuf) msg;
        log.info("|OutboundHandler2 write : "+data.toString(CharsetUtil.UTF_8));
        ctx.write(Unpooled.copiedBuffer("OutboundHandler2 "+data.toString(CharsetUtil.UTF_8), CharsetUtil.UTF_8));
        ctx.flush();
    }
}

　　InboundHandler1

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    private final static Logger log = LoggerFactory.getLogger(InboundHandler1.class);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        ByteBuf data = (ByteBuf) msg;
        log.info("|InboundHandler1 channelRead 服务端收到数据：" + data.toString(CharsetUtil.UTF_8));

        // 执行下一个InboundHandler
        ctx.fireChannelRead(Unpooled.copiedBuffer("InboundHandler1 " + data.toString(CharsetUtil.UTF_8), CharsetUtil.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

　　InboundHandler2

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    private final static Logger log = LoggerFactory.getLogger(InboundHandler2.class);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
        throws Exception {
        ByteBuf data = (ByteBuf) msg;
        log.info("|InboundHandler2 channelRead 服务端收到数据：" + data.toString(CharsetUtil.UTF_8));
        ctx.writeAndFlush(Unpooled.copiedBuffer("InboundHandler2 "+data.toString(CharsetUtil.UTF_8), CharsetUtil.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
        throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

• 客户端

　　客户端启动类

bootstrap.group(group)
	.channel(NioSocketChannel.class)
	.remoteAddress(new InetSocketAddress(host, port))
	.option(ChannelOption.TCP_NODELAY,true)
	.handler(new ChannelInitializer<SocketChannel>() {
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
			ch.pipeline().addLast(new EchoClientHandler());
		}
	});

　　

　　EchoClientHandler

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private final static Logger logger = LoggerFactory.getLogger(EchoServerHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {

        logger.info("Client received: " + msg.toString(CharsetUtil.UTF_8));

    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("Active");
        ctx.writeAndFlush(Unpooled.copiedBuffer("echo test...",CharsetUtil.UTF_8));
    }


    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        logger.info("EchoClientHandler channelReadComplete");
        ctx.close();
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

　　

执行结果如下：

两个OutboundHandler并没有执行，很奇怪，翻看源码，发现是InboundHandler的使用不对； 

pipeline上添加handler最终会往执行下面的流程；

io.netty.channel.DefaultChannelPipeline#addLast(io.netty.util.concurrent.EventExecutorGroup, java.lang.String, io.netty.channel.ChannelHandler)

io.netty.channel.DefaultChannelPipeline#addLast0

最终pipeline上的实例如下：

如下图

 

• 使用pipeline或channel发送数据

 将上面的InboundHandler2的channelRead方法中的发送数据方式改成如下方式：

ctx.pipeline().writeAndFlush(Unpooled.copiedBuffer("|InboundHandler2 "+data.toString(CharsetUtil.UTF_8), CharsetUtil.UTF_8));
ctx.channel().writeAndFlush(Unpooled.copiedBuffer("|InboundHandler2 "+data.toString(CharsetUtil.UTF_8), CharsetUtil.UTF_8));

　　

当在InboundHandler执行write方法，执行流程如下；

调用链路如下：

io.netty.channel.AbstractChannel#writeAndFlush(java.lang.Object)

 ->io.netty.channel.DefaultChannelPipeline#writeAndFlush(java.lang.Object)

  ->io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object)

   ->io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object, io.netty.channel.ChannelPromise)

    ->io.netty.channel.AbstractChannelHandlerContext#findContextOutbound

　　

io.netty.channel.DefaultChannelPipeline#writeAndFlush(java.lang.Object)

findContextOutbound会从tail节点往前找OutboundHandler，如下图；

最终的执行结果如下：

服务端：

客户端：

• 使用ctx发送数据的方式

而InboundHandler2的channelRead方法的发送数据是下面这种方式：

ctx.writeAndFlush(xxx);

与pipeline或channel的发送不同的是在执行 io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object, io.netty.channel.ChannelPromise) 前会执行io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object)；

此时ctx为InboundHandler2；

 

此时会从ctx往前找OutboundHandler，从pipeline上看ctx往前找是没有的，如下图；

 如果不想改变InboundHandler2的channelRead方法，可以将服务端的handler添加顺序更改就可以了，如下；

serverBootstrap.group(bossGroup, workGroup)
    .channel(NioServerSocketChannel.class)
    .option(ChannelOption.SO_BACKLOG,1024)
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
		    ch.pipeline().addLast(new OutboundHandler1());
            ch.pipeline().addLast(new OutboundHandler2());
            ch.pipeline().addLast(new InboundHandler1());
            ch.pipeline().addLast(new InboundHandler2());
        }
    });

执行效果如下：

服务端

客户端：

最终handler间的事件传播方向，也就是执行方向，InboundHandler之间传递数据，通过ctx.fireChannelRead(msg)，InboundHandler顺序执行，OutboundHandler逆序执行，如下；

传播的数据也会像包装一样，InboundHandler2会将InboundHandler1的数据打包，传递到OutboundHandler2，再打包，在传递到OutboundHandler1；感觉这个有点像StringBuilder的append方法；