理解ChannelPipeline

叁歲伎倆 2023-09-27 18:37 165阅读 0赞

### ChannelPipeline创建时机、Channel和ChannelPipeline的关系 ###

*  ChannelPipeline的创建时机，就是在调用bind方法时，会调用newChannel(),换句话说，创建Channel实例的时候会ChannelPipeline实例。
    
    ![format_png][]
 *  Channel和ChannelPipeline之间是互相关联的。
 *  一个Channel可以关联一个ChannelPipeline，而一个ChannelPipeline中又包含多个ChannelHandler，这些ChannelHandler可以处理Channel中的数据流。当Channel接收到数据时，会在ChannelPipeline中按照一定的顺序经过各个ChannelHandler的处理。

#### 理解ChannelPipeline ####

*  ChannelPipeline是由一系列(List)的ChannelHandler组成，用于处理或拦截Channel的入站事件和出站操作。它实现了一种[高级形式的Intercepting Filter模式 (oracle.com)][Intercepting Filter_ _oracle.com]，使得用户可以完全控制事件的处理方式以及在Pipeline中的各个ChannelHandler之间的交互方式。
    
     *  高级的地方，以Servlet的Filter为例
    
        public class CustomHeaderFilter implements Filter {
          @Override
          public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
                  throws IOException, ServletException {
        
              // 在响应中添加自定义 HTTP 头部
              HttpServletResponse httpServletResponse = (HttpServletResponse) response;
              httpServletResponse.addHeader("X-Custom-Header", "Hello World!");
        
              // 调用 FilterChain，继续处理请求和响应
              chain.doFilter(request, response);
          }
        } 
        复制代码
    
     *  可以发现request（入站），response（出站）走的都是doFilter方法，没有将出入站解耦。
     *  chain.doFilter(request, response);又表示拦截器是单向的，显然也不够灵活。
 *  当新的Channel创建时，Pipeline也会自动创建。
 *  Netty的 ChannelPipeline 是双向的，既可以处理入站事件也可以处理出站事件。将出入站给分开了。
 *  同时，Netty的ChannelPipeline支持动态修改，可以在运行时添加、删除、替换ChannelHandler，从而灵活地调整数据流的处理方式。
    
     *  Handler可以理解成Filter，但是Handler是事件触发的。

#### ChannelPipeline的工作流程 ####

![format_png 1][]

*  I/O 事件由ChannelInboundHandler或ChannelOutboundHandler处理，并通过调用ChannelHandlerContext中定义的事件传播方法（例如ChannelHandlerContext.fireChannelRead(Object)和ChannelHandlerContext.write(Object)**转发**到其最近的处理程序。
 *  入站事件是从底向上由入站处理器处理的，如左侧所示。入站处理器通常处理底部的I/O线程生成的入站数据。入站数据通常通过实际的输入操作（如SocketChannel.read(ByteBuffer)）从远程对等方读取。如果入站事件超出了顶部入站处理器，则会被静默丢弃，或者在需要您注意时记录日志。
    
     *  如果事件到达了顶部处理器，但没有被处理，那么这个事件就已经超出了顶部处理器的范围，因为顶部处理器是最上面的处理器。这就是所谓的“超出”。
     *  简单说就是数据到最顶层（最后一个）入站Handler，没有进一步的操作，那么该数据或者事件会被丢弃。
    
        public class MyInboundHandler1 extends ChannelInboundHandlerAdapter {
          @Override
          public void channelRead(ChannelHandlerContext ctx, Object msg) {
              // 处理入站事件
              // ...
              // 调用下一个入站处理器
              ctx.fireChannelRead(msg);
          }
        }
        
        public class MyInboundHandler2 extends ChannelInboundHandlerAdapter {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) {
              // 处理入站事件
              // ...
              // 不调用下一个入站处理器
              // 此时入站事件被处理并停止传递
              //如果没有后续操作，数据（事件）传递就到此为止了，
              //可以整个日志显示一下
              //或者，调用出站Handler，让事件继续跑
            }
        }
        
        // 在 ChannelPipeline 中添加处理器
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("handler1", new MyInboundHandler1());
        pipeline.addLast("handler2", new MyInboundHandler2());
        复制代码
 *  出站事件是从顶向下由出站处理器处理的，如右侧所示。出站处理器通常生成或转换出站流量，如写请求。如果出站事件超出了底部出站处理器，则会交由与Channel关联的I/O线程处理。I/O线程通常执行实际的输出操作，如SocketChannel.write(ByteBuffer)。
    
        public class MyOutboundHandler1 extends ChannelOutboundHandlerAdapter {
         @Override
         public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
             // 处理出站事件
             // ...
             // 调用下一个出站处理器
             ctx.write(msg, promise);
         }
        }
        
        public class MyOutboundHandler2 extends ChannelOutboundHandlerAdapter {
           @Override
           public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
             // 处理出站事件
             // ...
             // 不调用下一个出站处理器
             // 此时出站事件被处理并停止传递
             //会直接将其传入I/O线程执行SocketChannel.write(ByteBuffer).
           }
        }
        
        // 在 ChannelPipeline 中添加处理器
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("handler1", new MyOutboundHandler1());
        pipeline.addLast("handler2", new MyOutboundHandler2());
        复制代码
    
     *  没有调用下一个出站处理器，因此出站事件被处理并停止传递。如果出站事件超出了底部出站处理器，则会交由与 `Channel` 关联的 I/O 线程处理，执行实际的输出操作，如 `SocketChannel.write(ByteBuffer)`。

--------------------

*  ChannelPipeline是Netty中重要的概念之一，它是一系列的ChannelHandler的集合，用于处理传入和传出的数据。在这个pipeline中，数据会按照特定的顺序依次经过每个handler进行处理。每个handler会处理特定的任务，并将数据传递给下一个handler进行进一步处理。这样的数据处理过程类似于流水线加工，可以实现高效、灵活的数据处理。
 *  例如，假设我们创建了以下pipeline:
    
        ChannelPipeline p = ...;
        p.addLast("1", new InboundHandlerA());
        p.addLast("2", new InboundHandlerB());
        p.addLast("3", new OutboundHandlerA());
        p.addLast("4", new OutboundHandlerB());
        p.addLast("5", new InboundOutboundHandlerX());
        复制代码
    
     *  在上面的示例配置中，以Inbound开头的类表示它是一个入站处理器，以Outbound开头的类表示它是一个出站处理器。当事件进入时，处理器的评估顺序为1、2、3、4、5；当事件离开时，顺序为5、4、3、2、1。此外，为了缩短处理器的堆栈深度，ChannelPipeline会跳过某些处理器的评估：
     *  处理器3和4没有实现ChannelInboundHandler接口，因此实际的入站事件评估顺序为1、2、5。
     *  处理器1和2没有实现ChannelOutboundHandler接口，因此实际的出站事件评估顺序为5、4、3。
     *  如果处理器5同时实现了ChannelInboundHandler和ChannelOutboundHandler，则入站事件和出站事件的评估顺序分别为1、2、5和5、4、3。
     *  总之，InboundHandler上往下，OutboundHandler从从下往上 -> 1、2、5、5、4、3；
 *  为了将事件传递给下一个处理器，处理器**必须**调用ChannelHandlerContext中的事件传播方法。这些方法包括：
 *  入站事件传播方法：
    
     *  ChannelHandlerContext.fireChannelRegistered()
     *  ChannelHandlerContext.fireChannelActive()
     *  ChannelHandlerContext.fireChannelRead(Object)
     *  ChannelHandlerContext.fireChannelReadComplete()
     *  ChannelHandlerContext.fireExceptionCaught(Throwable)
     *  ChannelHandlerContext.fireUserEventTriggered(Object)
     *  ChannelHandlerContext.fireChannelWritabilityChanged()
     *  ChannelHandlerContext.fireChannelInactive()
     *  ChannelHandlerContext.fireChannelUnregistered()
 *  出站事件传播方法：
    
     *  ChannelHandlerContext.bind(SocketAddress, ChannelPromise)
     *  ChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
     *  ChannelHandlerContext.write(Object, ChannelPromise)
     *  ChannelHandlerContext.flush()
     *  ChannelHandlerContext.read()
     *  ChannelHandlerContext.disconnect(ChannelPromise)
     *  ChannelHandlerContext.close(ChannelPromise)
     *  ChannelHandlerContext.deregister(ChannelPromise)
 *  以下是一个事件传播的示例：
    
        public class MyInboundHandler extends ChannelInboundHandlerAdapter {
            @Override
            public void channelActive(ChannelHandlerContext ctx) {
                System.out.println("Connected!");
                ctx.fireChannelActive();
            }
        }
        
        public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {
            @Override
            public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
                System.out.println("Closing ..");
                ctx.close(promise);
            }
        }
        复制代码
 *  构建管道
    
     *  用户应该在管道中拥有一个或多个ChannelHandler来接收I/O事件（例如读取）并请求I/O操作（例如写入和关闭）。例如，一个典型的服务器将在每个通道的管道中拥有以下处理程序，但根据协议和业务逻辑的复杂性和特性，您的情况可能会有所不同：
     *  协议解码器 - 将二进制数据（例如ByteBuf）转换为Java对象。
     *  协议编码器 - 将Java对象转换为二进制数据。
     *  业务逻辑处理程序 - 执行实际的业务逻辑（例如数据库访问）。
     *  可以通过以下示例表示：
    
        static final EventExecutorGroup group = new DefaultEventExecutorGroup(16); ...
        
        ChannelPipeline pipeline = ch.pipeline();
        
        pipeline.addLast("decoder", new MyProtocolDecoder()); 
        pipeline.addLast("encoder", new MyProtocolEncoder());
        
        // 告诉管道在不同的线程中运行MyBusinessLogicHandler的事件处理程序方法，
        // 以便I/O线程不会被耗时的任务阻塞。 
        // 如果您的业务逻辑完全是异步的或完成非常快，您不需要指定一个组。   
        pipeline.addLast(group, "handler", new MyBusinessLogicHandler());
        //业务Handler应该自行用线程池异步执行
        复制代码
    
     *  请注意，虽然使用DefaultEventLoopGroup将从EventLoop中卸载操作，但它仍会按照每个ChannelHandlerContext的顺序以串行方式处理任务，并保证排序。由于顺序，它可能仍然成为一个瓶颈。如果您的用例不需要排序，您可能希望考虑使用UnorderedThreadPoolEventExecutor以最大化任务执行的并行性。
        
            EventLoopGroup group = new DefaultEventLoopGroup();
            Channel channel = new SomeChannel();
            channel.pipeline().addLast(group, "handler", new SomeHandler());
            复制代码
        
         *  上述代码中，group被用作ChannelPipeline的执行器。因此，如果需要维护任务执行的顺序，则使用DefaultEventLoopGroup是个不错的选择。
        
            EventExecutor executor = new UnorderedThreadPoolEventExecutor(10);
            Channel channel = new SomeChannel();
            channel.pipeline().addLast(executor, "handler", new SomeHandler());
            复制代码
        
         *  上述代码中，UnorderedThreadPoolEventExecutor将以无序方式执行任务，这意味着任务将在不同的线程中同时执行，以最大化执行的并行性。
     *  默认情况下，**处理器是在与事件相关的I/O线程上执行的**。但是，如果处理器需要执行一些耗时的任务，它可能会导致I/O线程阻塞，从而影响性能。
 *  线程安全
    
     *  ChannelPipeline是线程安全的，因此可以随时添加或删除ChannelHandler。例如，在敏感信息即将交换时，可以插入加密处理程序，并在交换后删除它。
    
        public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
        
          private static final String ENCRYPTION_HANDLER_NAME = "encryptionHandler";
          private static final String BUSINESS_HANDLER_NAME = "businessHandler";
        
          @Override
          protected void initChannel(SocketChannel ch) throws Exception {
              ChannelPipeline pipeline = ch.pipeline();
              pipeline.addLast(BUSINESS_HANDLER_NAME, new BusinessHandler());
        
              // 加密处理程序在敏感信息即将交换时插入
              pipeline.addBefore(BUSINESS_HANDLER_NAME, ENCRYPTION_HANDLER_NAME, new EncryptionHandler());
        
              // 在交换后删除加密处理程序
              pipeline.remove(ENCRYPTION_HANDLER_NAME);
          }
        }
        复制代码
 *  入站事件处理器如何转到出站处理器的：
    
        public class MyInboundHandler extends ChannelInboundHandlerAdapter {
        
          @Override
          public void channelRead(ChannelHandlerContext ctx, Object msg) {
              // 接收来自客户端的消息
              String message = (String) msg;
        
              // 将消息编码为特定的协议
              byte[] encodedMessage = MyProtocol.encode(message);
        
              // 将消息发送回客户端
              ctx.write(encodedMessage);//从入站转移到了出站
              ctx.flush(); // 在最后一次调用write()方法后刷新缓冲区
          }
        }
        
        public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {
        
          @Override
          public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
              // 修改消息或发送其他事件
              byte[] encodedMessage = (byte[]) msg;
              byte[] modifiedMessage = MyProtocol.modify(encodedMessage);
        
              // 发送修改后的消息
              ctx.write(modifiedMessage, promise);
          }
        }
        复制代码
    
     *  在上面的示例中，`MyInboundHandler`接收来自客户端的消息，并将其编码为特定的协议。然后，它将消息通过`write()`方法发送给下一个`ChannelOutboundHandler`，即`MyOutboundHandler`。`MyOutboundHandler`修改了消息并将其发送回客户端。使用这种方式，您可以在入站和出站处理程序之间有效地传递事件，并根据需要对其进行修改。

#### 小结 ####

*  ChannelPipeline实际上是一个事件处理器链，负责处理传入或传出的事件。

作者：KittyGuy  
链接：https://juejin.cn/post/7228949826607071287

[format_png]: https://img-blog.csdnimg.cn/img_convert/3036081f2aa2edcae0b1c37c5e58c397.webp?x-oss-process=image/format,png
[Intercepting Filter_ _oracle.com]: https://link.juejin.cn?target=https%3A%2F%2Fwww.oracle.com%2Fjava%2Ftechnologies%2Fintercepting-filter.html
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/28eaa2e46fb89f4bf8bef41ef843ec57.webp?x-oss-process=image/format,png