Netty ChannelPipeline 数据管道

今天药忘吃喽~ 2022-10-30 12:20 188阅读 0赞

# 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。

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

![图片描述][334be90aae21dbfda948c43f0c7fed18.png]

## 3. ChannelPipeline 核心方法 ##

ChannelPipeline 的最常用方法：

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

实例：

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

[334be90aae21dbfda948c43f0c7fed18.png]: /images/20221024/5f08e72d06e2407f9303ab8a5a130a17.png