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 的最常用方法:

方法 描述
addFirst(…) 添加 ChannelHandler 在 ChannelPipeline 的第一个位置
addBefore(…) 在 ChannelPipeline 中指定的 ChannelHandler 名称之前添加 ChannelHandler
addAfter(…) 在 ChannelPipeline 中指定的 ChannelHandler 名称之后添加 ChannelHandler
addLast(…) 在 ChannelPipeline 的末尾添加 ChannelHandler
remove(…) 删除 ChannelPipeline 中指定的 ChannelHandler
replace(…) 替换 ChannelPipeline 中指定的 ChannelHandler
ChannelHandler first() 获取链表当中的第一个节点
ChannelHandler last() 获取链表当中的最后一个节点

实例:

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. 入站和出站的流转顺序和业务逻辑的执行顺序。