高性能NIO通信框架之Netty（3）ChannelPipeline分析

刺骨的言语ヽ痛彻心扉 2021-10-18 18:56 393阅读 0赞

一．ChannelPipeline和ChannelHandler的简介

Netty的ChannelPipeline和ChannelHandler机制类似于Servlet和Filter过滤器，这类拦截器实际上是职责责任链模式的一种变形，主要是为了方便事件的拦截和用户业务逻辑的定制。

Netty的Channel过滤器实现原理与Servlet Filter机制一致，它将Channel的数据管道抽象为ChannelPipeline，消息ChannelPipeline中流动和传递。ChannelPipeline持有I/O事件拦截器ChannelPipeline的链表，由ChannelHandler对I/O事件进行拦截和处理，可以方便的通过新增和删除ChannelHandler来实现不同的业务逻辑定制，不需要对已有的ChannelHandler进行修改，能够实现对修改封闭和对扩展的支持。

二．ChannelPipeline的功能说明

ChannelPipeline是ChannelHandler的容器，它负责ChannelHandler的管理和事件拦截与调度。

![1166389-20190714170010264-1124729201.png][]

上图展示了一个消息被ChannelPipeline的ChannelHandler链拦截和处理的全过程，消息的读取和发送处理全流程描述如下：

（1）底层的SocketChannel read()方法读取ByteBuf，触发ChannelRead事件，由I/O线程NioEventLoop调用ChannelPipeline的fireChannelRead(Object msg)方法，将ByteBuf消息传输到ChannelPipeline中。

（2）消息依次被HeadHandler、ChannelHandler1、ChannelHandler2 .... ChannelHandler N-1  ChannelHandler N  TailHandler拦截和处理，在这个过程中，任何ChannelHandler都可以中断当前的流程，结束消息的传递。

（3）调用ChannelHandlerContext的write方法发送消息，消息从TailHandler开始经过ChannelHandlerN ..... ChannelHandler1、HeadHandler，最终被添加到消息发送缓冲区中等待刷新和发送，在此过程中也可以中断消息的传递，例如当编码失败时，就需要中断流程，构造异常的Future返回。

Netty中的事件分为inbound和outbound事件，inbound事件通常由I/O线程触发，例如TCP链路建立事件、链路关闭事件、读事件、异常通知事件等，接下来介绍下inbound事件和outbound事件，主要在ChannelHandlerContext里面：

触发Inbound事件的方法如下：

（1）fireChannelRegistered():Channel注册事件

（2）fireChannelActive():TCP链路建立成功，Channel激活事件

（3）fireChannelRead(Object):读事件

（4）fireChannelReadComplete()：读操作完成通知事件

（5）fireExceptionCaught(Throwable)：异常事件通知

（6）fireUserEventTriggered(Object)：用户自定义事件

（7）fireChannelWritabilityChanaged()：Channel的可写状态变化通知事件

（8）fireChannelInactive()：TCP链路关闭，链路不可用通知事件

outBound事件通常是由用户主动发起的网络I/O操作，例如用户发起的连接操作、绑定操作、消息发送等操作

触发outbound事件的方法如下：

（1）bind(SocketAddress,ChannelPromise):绑定本地地址事件

（2）connect(SocketAddress,SocketAddress,ChannelPromise)：连接服务端事件

（3）write(Object,ChannelPromise)：发送事件

（4）flush()：刷新事件

（5）read()：读事件

（6）disconnect(ChannelPromise):断开连接事件

（7）close(ChannelPromise)：关闭当前Channel事件

2.1 自定义拦截器

ChannelPipeline通过ChannelHandler接口实现事件的拦截和处理，由于ChannelHandler中的事件种类繁多，不同的ChannelHandler可能只需要关心其中的某一个或者几个事件，所以通常ChannelHandler只需要继承ChannelHandlerAdapter类覆盖自己关心的方法即可；例如：拦截ChannelActive事件，打印TCP链路建立成功日志，代码如下：

public class LogPrintChannelHandler extends ChannelHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { //打印相关的日志
            System.out.println("TCP Connected"); //然后将事件继续传递给下一个ChannelHandler
     ctx.fireChannelActive(); } }

2.2  ChannelPipeline的重要特性之动态增加和删除ChannelHandler

ChannelPipeline支持运行状态动态的添加和删除ChannelHandler，在某些场景下这个特性非常实用。例如当业务高峰期需要对系统做拥塞保护时，就可以根据当前的系统时间进行判断，如果处于业务高峰期，则动态地将系统拥塞保护ChannelHandler添加到当前的ChannelPipeline中，当高峰期过去之后，就可以动态删除拥塞保护ChannelHandler了。

ChannelPipeline是线程安全的，这意味着N个业务线程可以并发地操作ChannelPipeline而不存在多线程并发问题，这意味着尽管ChannelPipeline是线程安全的，但是用户仍然需要自己保证ChannelHandler的线程安全。

三．ChannelPipeline源码分析

ChannelPipeline的代码相对比较简单，它实际上是一个ChannelHandler的容器，内部维护了一个ChannelHandler的链表和迭代器，可以方便地实现ChannelHandler查找、添加、替换和删除。

ChannelPipeline的类继承关系如下：

![1166389-20190714170114550-801695984.png][]

3.1 ChannelPipeline对ChannelHandler的管理

ChannelPipeline是ChannelHandler的管理容器，负责ChannelHandler的查询、添加、替换和删除。由于它与Map等容器的实现非常相似。接口代码如下：

![1166389-20190714170127404-675301813.png][]

由于ChannelPipeline支持动态运行期间动态修改，因此存在两种潜在的多线程并发访问场景：

1）I/O线程和用户业务线程的并发访问

2）用户多个线程之间的并发访问

同时为了保证ChannelPipeline的线程安全性，需要通过线程安全容器或者锁来保证并发操作的安全，Netty使用了synchronized关键字，保证同步块内的所有操作的原子性。首先根据baseName获取它对应的DefaultChannelHandlerContext，ChannelPipeline维护了ChannelHandler名称和ChannelHandlerContext实例的映射关系，代码如下：

public ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler) { return this.addBefore((ChannelHandlerInvoker)null, baseName, name, handler); } public ChannelPipeline addBefore(EventExecutorGroup group, String baseName, String name, ChannelHandler handler) { synchronized(this) { AbstractChannelHandlerContext ctx = this.getContextOrDie(baseName); name = this.filterName(name, handler); this.addBefore0(name, ctx, new DefaultChannelHandlerContext(this, this.findInvoker(group), name, handler)); return this; } }

public ChannelHandlerContext context(String name) { if(name == null) { throw new NullPointerException("name"); } else { synchronized(this) { return (ChannelHandlerContext)this.name2ctx.get(name); } } }

对新增的ChannelHandler名进行重复性校验，如果已经有同名的ChannelHandler存在，则不允许覆盖，抛出IllegalArgumentException("Duplicate handler name: " + name);异常。校验通过之后，使用新增的ChannelHandler等参数构造一个新的DefaultChannelHandlerContext实例；

private String filterName(String name, ChannelHandler handler) { if(name == null) { return this.generateName(handler); } else if(!this.name2ctx.containsKey(name)) { return name; } else { throw new IllegalArgumentException("Duplicate handler name: " + name); } }

将新建的DefaultChannelHandlerContext添加到当前的pipeline中，

private void addBefore0(String name, AbstractChannelHandlerContext ctx, AbstractChannelHandlerContext newCtx) { checkMultiplicity(newCtx); newCtx.prev = ctx.prev; newCtx.next = ctx; ctx.prev.next = newCtx; ctx.prev = newCtx; this.name2ctx.put(name, newCtx); this.callHandlerAdded(newCtx); }

如上代码，在添加之前需要对ChannelHandlerContext做重复性校验，如果ChannelHandlerContext不是可以在多个ChannelPipeline中共享的，且已经被添加到ChannelPipeline中则抛出ChannelPipelineException异常，具体代码如下：

private static void checkMultiplicity(ChannelHandlerContext ctx) { ChannelHandler handler = ctx.handler(); if(handler instanceof ChannelHandlerAdapter) { ChannelHandlerAdapter h = (ChannelHandlerAdapter)handler; if(!h.isSharable() && h.added) { throw new ChannelPipelineException(h.getClass().getName() + " is not a @Sharable handler, so can't be added or removed multiple times."); } h.added = true; } }

ChannelPipeline本身并不直接进行I/O操作，在前面对Channel和Unsafe的介绍中我们知道最终都是由Unsafe和Channel来实现真正的I/O操作的。Pipeline负责将I/O事件通过TailHandler进行调度和传播，最终调用Unsafe的I/O方法进行I/O操作，然后直接调用TailHandler 的connect方法，最终会调用到HeadHandler的connect方法，最终由HeadHandler调用Unsafe的connect方法发起真正的连接，pipeline仅仅负责事件的调度。

转载于:https://www.cnblogs.com/lovegrace/p/11184813.html

[1166389-20190714170010264-1124729201.png]: /images/20211018/a1121c677855450985dcbbae21e9c0e6.png
[1166389-20190714170114550-801695984.png]: /images/20211018/262f17bb6814456497c20470b9000935.png
[1166389-20190714170127404-675301813.png]: /images/20211018/745fbca0f5f249acbec47285f6303032.png