ChannelPipeline 源码分析（二十三）

怼烎@ 2022-10-29 14:20 197阅读 0赞

今天开始分析Netty源码：ChannelPipeline 源码分析

**深入**  **Netty**

从前面对  Netty  的使用我们看到，在运行的时候， Netty  中的组件一起为基于  Netty  的网络通信做出了贡献，所以下面我们来一一分析 Netty  中的组件，包括  ChannelPipeline 、 ChannelHandlerContext、 ChannelHandler  、 EventLoop 、 Channel 、 Unsafe 、 ByteBuf  等等，并且了解当服务器启动、当各种事件发生时 Netty  内部的运行机制。

**一、ChannelPipeline**  **源码分析 ** **简介**

Netty  的  ChannelPipeline  和  ChannelHandler  机制类似于  Servlet  和  Filter  过滤器，这类拦截器实际上是职责链模式的一种变形，主要是为了方便事件的拦截和用户业务逻辑的定制。Servlet Filter 是  J2EE Web  应用程序级的  Java  代码组件，它能够以声明的方式插入到HTTP 请求响应的处理过程中，用于拦截请求和响应，以便能够查看、提取或以某种方式操 作正在客户端和服务器之间交换的数据。拦截器封装了业务定制逻辑，能够实现对 Web  应用程序的预处理和事后处理。过滤器提供了一种面向对象的模块化机制，用来将公共任务封装到可插入的组件中。这 些组件通过 Web  部署配置文件  ( web.xml ）进行声明，可以方便地添加和删除过滤器，无须改动任何应用程序代码或 JSP  页面，由  Servlet  进行动态调用。通过在请求 / 响应链中使用过 滤器，可以对应用程序（而不是以任何方式替代）的 Servlet  或  JSP  页面提供的核心处理进行补充，而不破坏 Servlet  或  JSP  页面的功能。由于是纯  Java  实现，所以  Servlet  过滤器具有跨平台的可重用性，使得它们很容易被部署到任何符合 Servlet  规范的  J2EE  环境中。Netty 的  Channel  过滤器实现原理与  Servlet Filter  机制一致，它将  Channel  的数据管道抽象为 ChannelPipeline ，消息在  ChannelPipeline  中流动和传递。 ChannelPipeline  持有  IO  事件拦截器 ChannelHandler  的链表，由  ChannelHandler  对  IO  事件进行拦截和处理，可以方便地通过新增和删除 ChannelHandler  来实现不同的业务逻辑定制，不需要对已有的ChannelHandler 进行修改，能够实现对修改封闭和对扩展的支持。

**二、ChannelPipeline**  **的事件处理**

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

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

(2 ）消息依次被  ChannelPipeline  中的各个入站  Handler  拦截和处理，在这个过程中，任何 ChannelHandler  都可以中断当前的流程，结束消息的传递。

(3 ）调用  write  方法发送消息，消息被出站  Handler  拦截和处理，最终被添加到消息发送缓冲区中等待刷新和发送，在此过程中也可以中断消息的传递，例如当编码失败时，就需要中断流程，构造异常的 Future  返回。

Netty  中的事件分为  inbound  事件和  outbound  事件。 inbound  事件通常由  IO  线程触发， 例如 TCP  链路建立事件、链路关闭事件、读事件、异常通知事件等。

2、触发  inbound  事件的方法如下：

(1) ChannelHandlerContext.fireChannelRegistered(): Channel  注册事件 ;

(2)ChannelHandlerContext.fireChannelActive():TCP  链路建立成功， Channel  激活事件 ;

( 3)ChannelHandlerContext.fireChannelRead(Object): 读事件 ;

(4) ChannelHandlerContext.fireChannelReadComplete(): 读操作完成通知事件 ;

( 5) ChannelHandlerContext.fireExceptionCaught(Throwable): 异常通知事件 ;

( 6) ChannelHandlerContext.fireUserEventTriggered(Object):用户自定义事件 :

( 7)ChannelHandlerContext.fireChannelWritabilityChanged(): Channel  的可写状态变化通知事件;

( 8)ChannelHandlerContext.fireChannelInactive():TCP  连接关闭，链路不可用通知事件。

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

3、触发  outbound  事件的方法如下 :

( 1) ChanneHandlerContext.bind(Socket.Address, ChannelPromise): 绑定本地地址事件 ; ( 2)ChannelHandlerContext.connect(SocketAddress ， SocketAddress ，  ChannelPromise): 连接服务端事件;

( 3 ） ChannelHandlerContext.write(Object, ChannelPromise): 发送事件 ;

( 4)ChannelHandlerContext.flush(): 刷新事件 ;

(5)ChannelHandlerContext.read(): 读事件 ;

(6 ） ChannelHandlerContext.disconnect(ChannelPromise): 断开连接事件；

(7)ChannelHandlerContext.close(ChannelPromise): 关闭当前  Channel  事件。

**三、构建**  **pipeline**  **和特性**

根据我们在前面对  Netty  的使用，我们知道，我们不需要自己创建  pipeline ，因为使用ServerBootstrap 或者  Bootstrap  启动服务端或者客户端时， Netty  会为每个  Channel  连接创建一个独立的 pipeline 。对于使用者而言，只需要将自定义的拦截器加入到  pipeline  中即可。ChannelPipeline 支持运行态动态的添加或者删除  ChannelHandler ，在某些场景下这个特性非常实用。例如当业务高峰期需要对系统做拥塞保护时，就可以根据当前的系统时间进行判断，如果处于业务高峰期，则动态地将系统拥塞保护 ChannelHandler 添加到当前的ChannelPipeline 中，当高峰期过去之后，就可以动态删除拥塞保护  ChannelHandler  了。ChannelPipeline 是线程安全的 , 这意味着  N  个业务线程可以并发地操作  ChannelPipeline 而不存在多线程并发问题。但是，ChannelHandler  却不是线程安全的，这意味着尽管ChannelPipeline 是线程安全的，但是用户仍然需要自己保证  ChannelHandler  的线程安全。

**1、类关系图**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

**2、DefaultChannelPipeline**

可以看见， ChannelPipeline  的实现其实只有一个  DefaultChannelPipeline ，所以我们对ChannelPipeline 的了解就集中在这个类上。

***3、对***  ***ChannelHandler***  ***的管理***

ChannelPipeline  是  ChannelHandler  的管理容器，必然要负责  ChannelHandler  的查询、添加、替换和删除，所以它里面的方法很多都是和管理 ChannelHandler  有关的，比如  add 系列、remove  系列、 replace  系列。 ChannelPipeline 内部维护了一个  ChannelHandler  的链表和迭代器，所以从本质上来说， 上述的管理操作基本上就是对链表的操作。比如add 操作源码：

@Override
    public final ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            checkMultiplicity(handler);
            name = filterName(name, handler);
    
            newCtx = newContext(group, name, handler);
    
            addFirst0(newCtx);
    
            // If the registered is false it means that the channel was not registered on an eventloop yet.
            // In this case we add the context to the pipeline and add a task that will call
            // ChannelHandler.handlerAdded(...) once the channel is registered.
            if (!registered) {
                newCtx.setAddPending();
                callHandlerCallbackLater(newCtx, true);
                return this;
            }
    
            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {
                newCtx.setAddPending();
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        callHandlerAdded0(newCtx);
                    }
                });
                return this;
            }
        }
        callHandlerAdded0(newCtx);
        return this;
    }

但是 ChannelHandler 本身并不是链表上的元素，所以在具体实现上是把 ChannelHandler 包装进 ChannelHandlerContext 的实例 DefaultChannelHandlerContext，然后把 ChannelHandlerContext 作为元素来组成链表。 由于 ChannelPipeline 支持运行期动态修改，因此存在两种潜在的多线程并发访问场景。

1) 、 IO  线程和用户业务线程的并发访问 ;

2) 、用户多个线程之间的并发访问。

为了保证  ChannelPipeline  的线程安全性，需要通过线程安全容器或者锁来保证并发访问的安全，此处 Netty  直接使用了  synchronized  关键字，保证同步块内的所有操作的原子性。在加入链表之前，Netty  还会检查该  Handler  是否已经被添加过及其名字是否有重复，如果该 Handler  不是共享的，而且被添加过抛出异常，如果名字重复，也会抛出异常。 完成链表操作后，后面的部分基本上做的就是一件事，执行方法 callHandlerAdded0 ，只是根据条件不同进行同步或者异步执行，callHandlerAdded0  的主要作用是在  handler  添加被加入链表之后做一些额外工作，Netty  本身对这个方法的实现，在  ChannelHandlerAdapter  类中，是个空操作

@Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // NOOP
    }

这里如果我们需要做一些业务逻辑, 可以通过重写该方法进行实现。

***4、inbound***  ***系列***

当发生某个  I/O  事件的时候，例如链路建立、链路关闭、读取操作完成等，都会产生一个事件，事件在 pipeline  中得到传播和处理，它是事件处理的总入口。由于网络  IO  相关的事件有限, 因此 Netty 对这些事件进行了统一抽象 , Netty 自身和用户的 ChannelHandler 会对感兴趣的事件进行拦截和处理。pipeline 中以  fireXXX  命名的方法都是从  IO  线程流向用户业务  Handler  的  inbound  事件，它们的实现因功能而异，但是处理步骤类似。

总结如下：

(1) 调用  HeadHandler  对应的  fireXXX  方法 ;

(2 ）执行事件相关的逻辑操作。

***5、outbound***  ***系列***

由用户线程或者代码发起的  IO  操作被称为  outbound  事件 , 事实上  inbound  和  outbound 是 Netty  自身根据事件在  pipeline  中的流向抽象出来的术语，在其他  NIO  框架中并没有这个概念。outbound  事件包括发起绑定、发起连接、发起读写、刷新数据、发起关闭、发起断连等等。Pipeline 本身并不直接进行  IO  操作，最终都是由  Unsafe  和  Channel  来实现真正的  IO  操作的。Pipeline  负责将  IO  事件通过  TailHandler  进行调度和传播，最终调用  Unsafe  的  IO  方法进行 I/O  操作。

到此，ChannelPipeline 源码分析分析完毕，下篇分析ChannelHandlerContext相关源码，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/4a822f86cbcd422385bf4491772bec53.png