Netty——ChannelHandler和ChannelPipeline-白红宇

Netty——ChannelHandler和ChannelPipeline

阅读量：2444 次

发布时间：2019-05-10

本文共 6590 字，大约阅读时间需要 21 分钟。

ChannelHandler

Channel的生命周期：

Interface Channel 定义了一组和 ChannelInboundHandler API 密切相关的简单但功能强大的状态模型，下表列出了 Channel 的这 4 个状态：

当这些状态发生改变时，将会生成对应的事件。这些事件将会被转发给 ChannelPipeline 中的 ChannelHandler，其可以随后对它们做出响应。

ChannelHandler的生命周期：

在 ChannelHandler 被添加到 ChannelPipeline 中或者被从 ChannelPipeline 中移除时会调用这些操作。这些方法中的每一个都接受一个 ChannelHandlerContext 参数。

Netty 定义了下面两个重要的 ChannelHandler 子接口:

ChannelInboundHandler——处理入站数据以及各种状态变化;

ChannelOutboundHandler——处理出站数据并且允许拦截所有的操作。

下表列出了 interface ChannelInboundHandler 的生命周期方法。这些方法将会在数据被接收时或者与其对应的 Channel 状态发生改变时被调用。正如我们前面所提到的，这些方法和 Channel 的生命周期密切相关。

当某个 ChannelInboundHandler 的实现重写 channelRead()方法时，它将负责显式地释放与池化的 ByteBuf 实例相关的内存。Netty 为此提供了一个实用方法 ReferenceCountUtil.release()：

//释放消息资源@Sharablepublic class DiscardHandler extends ChannelInboundHandlerAdapter {
   	@Over	public void channelRead(ChannelHandlerContext ctx, Object msg){
   	//丢弃已接收的消息		ReferenceCountUtil.release(msg);	}}

Netty 将使用 WARN 级别的日志消息记录未释放的资源，使得可以非常简单地在代码中发现违规的实例。但是以这种方式管理资源可能很繁琐。一个更加简单的方式是使用 SimpleChannelInboundHandler:

@Sharablepublic class SimpleDiscardHandler extends SimpleChannelInboundHandler

由于 SimpleChannelInboundHandler 会自动释放资源，所以你不应该存储指向任何消息的引用供将来使用，因为这些引用都将会失效。

ChannelOutboundHandler接口：

出站操作和数据将由 ChannelOutboundHandler 处理。它的方法将被 Channel、ChannelPipeline 以及 ChannelHandlerContext 调用。

ChannelOutboundHandler 的一个强大的功能是可以按需推迟操作或者事件，这使得可以通过一些复杂的方法来处理请求。例如，如果到远程节点的写入被暂停了，那么你可以推迟冲刷操作并在稍后继续。

在这里插入图片描述

ChannelOutboundHandler中的大部分方法都需要一个

ChannelPromise参数，以便在操作完成时得到通知。

ChannelPromise是ChannelFuture的一个 子类，其定义了一些可写的方法，如setSuccess()和setFailure()，从而使ChannelFuture不可变。

ChannelHandler适配器：

你可以使用 ChannelInboundHandlerAdapter 和 ChannelOutboundHandlerAdapter 类作为自己的 ChannelHandler 的起始点。这两个适配器分别提供了 ChannelInboundHandler 和 ChannelOutboundHandler 的基本实现。通过扩展抽象类 ChannelHandlerAdapter，它们获得了它们共同的超接口 ChannelHandler 的方法。生成的类的层次结构如图所示:

ChannelHandlerAdapter 还提供了实用方法 isSharable()。如果其对应的实现被标注为 Sharable，那么这个方法将返回 true，表示它可以被添加到多个 ChannelPipeline 中。

在 ChannelInboundHandlerAdapter 和 ChannelOutboundHandlerAdapter 中所提供的方法体调用了其相关联的 ChannelHandlerContext 上的等效方法，从而将事件转发到了 ChannelPipeline 中的下一个 ChannelHandler 中。

你要想在自己的 ChannelHandler 中使用这些适配器类，只需要简单地扩展它们，并且重写那些你想要自定义的方法。

资源管理：

每当通过调用 ChannelInboundHandler.channelRead()或者 ChannelOutboundHandler.write()方法来处理数据时，你都需要确保没有任何的资源泄漏。Netty 使用引用计数来处理池化的 ByteBuf。所以在完全使用完某个 ByteBuf 后，调整其引用计数是很重要的。

为了帮助你诊断潜在的(资源泄漏)问题，Netty提供了class ResourceLeakDetector1，它将对你应用程序的缓冲区分配做大约 1%的采样来检测内存泄露。相关的开销是非常小的。

Netty目前定义了4中泄漏检测级别：

泄露检测级别可以通过将下面的 Java 系统属性设置为表中的一个值来定义:

java -Dio.netty.leakDetectionLevel=ADVANCED

//消费并释放入站消息@ChannelHandler.Sharable    public class DiscardInboundHandler extends ChannelInboundHandlerAdapter {
           @Override        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
               //通过调用ReferenceCountUtil.realease()方法释放资源            ReferenceCountUtil.release(msg);        }    }

消费入站消息的简单方式由于消费入站数据是一项常规任务，所以 Netty 提供了一个特殊的被称为 SimpleChannelInboundHandler 的 ChannelInboundHandler 实现。这个实现会在消息被 channelRead0()方法消费之后自动释放消息。

在出站方向这边，如果你处理了 write()操作并丢弃了一个消息，那么你也应该负责释放它。

//丢弃并释放出栈消息@ChannelHandler.Sharable    public class DiscardOutboundHandler extends ChannelOutboundHandlerAdapter {
           @Override        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
               ReferenceCountUtil.release(msg);            //通知ChannelPromise数据已经被处理            promise.setSuccess();        }    }

重要的是，不仅要释放资源，还要通知 ChannelPromise。否则可能会出现 ChannelFutureListener 收不到某个消息已经被处理了的通知的情况。

总之，如果一个消息被消费或者丢弃了，并且没有传递给 ChannelPipeline 中的下一个 ChannelOutboundHandler，那么用户就有责任调用 ReferenceCountUtil.release()。 如果消息到达了实际的传输层，那么当它被写入时或者 Channel 关闭时，都将被自动释放。

ChannelPipeline接口

如果你认为 ChannelPipeline 是一个拦截流经 Channel 的入站和出站事件的 ChannelHandler 实例链，那么就很容易看出这些 ChannelHandler 之间的交互是如何组成一个应用程序数据和事件处理逻辑的核心的。

每一个新创建的 Channel 都将会被分配一个新的 ChannelPipeline。这项关联是永久性的;Channel 既不能附加另外一个 ChannelPipeline，也不能分离其当前的。在 Netty 组件的生命周期中，这是一项固定的操作，不需要开发人员的任何干预。

根据事件的起源，事件将会被 ChannelInboundHandler 或者 ChannelOutboundHandler处理。随后，通过调用 ChannelHandlerContext 实现，它将被转发给同一超类型的下一个ChannelHandler。

ChannelHandlerContext：

ChannelHandlerContext使得ChannelHandler能够和它的ChannelPipeline以及其他的 ChannelHandler 交互。 ChannelHandler 可以通知其所属的 ChannelPipeline 中的下一个 ChannelHandler，甚至可以动态修改它所属的ChannelPipeline。

下图展示了一个典型的同时具有入站和出站 ChannelHandler 的 ChannelPipeline 的布局，并且印证了我们之前的关于 ChannelPipeline 主要由一系列的 ChannelHandler 所组成的说法。C h a n n e l P i p e l i n e 还提供了通过 C h a n n e l P i p e l i n e 本身传播事件的方法。如果一个入站事件被触发，它将被从 ChannelPipeline 的头部开始一直被传播到 Channel Pipeline 的尾端。一个出站 I/O 事件将从 ChannelPipeline 的最右边开始，然后向左传播。

在这里插入图片描述

从事件途经 ChannelPipeline 的角度来看，ChannelPipeline 的头部和尾端取决于该事件是入站的还是出站的。然而 Netty 总是将 ChannelPipeline 的入站口(图中的左侧) 作为头部，而将出站口(该图的右侧)作为尾端。当你完成了通过调用 ChannelPipeline.add*()方法将入站处理器(ChannelInboundHandler) 和出站处理器(ChannelOutboundHandler)混合添加到 ChannelPipeline 之后，每一个 ChannelHandler 从头部到尾端的顺序位置正如同我们方才所定义它们的一样。因此，如果你将图中的处理器(ChannelHandler)从左到右进行编号，那么第一个被入站事件看到的 ChannelHandler 将是 1，而第一个被出站事件看到的 ChannelHandler 将是 5。

在 ChannelPipeline 传播事件时，它会测试 ChannelPipeline 中的下一个 ChannelHandler 的类型是否和事件的运动方向相匹配。如果不匹配，ChannelPipeline 将跳过该 ChannelHandler 并前进到下一个，直到它找到和该事件所期望的方向相匹配的为止。(当然，ChannelHandler 也可以同时实现 ChannelInboundHandler 接口和 ChannelOutboundHandler 接口。)

修改ChannelPipeline：

ChannelHandler 可以通过添加、删除或者替换其他的 ChannelHandler 来实时地修改 ChannelPipeline 的布局。(它也可以将它自己从 ChannelPipeline 中移除。)这是 Channel- Handler 最重要的能力之一：

ChannelPipeline pipeline = null;        FirstHandler firstHandler = new FirstHandler();        pipeline.addLast("handler1", firstHandler);        pipeline.addFirst("handler2", new SecondHandler());        pipeline.addLast("handler3", new ThirdHandler());        //此时pipeline为：2 1 3        pipeline.remove("handler3");//通过名称移除        pipeline.remove(firstHandler);//通过引用移除        pipeline.replace("handler2", "handler4", new ForthHandler());        //此时pipeline为：4

ChannelHandler的执行和阻塞：

通常 ChannelPipeline 中的每一个 ChannelHandler 都是通过它的 EventLoop(I/O 线程)来处理传递给它的事件的。所以至关重要的是不要阻塞这个线程，因为这会对整体的 I/O 处理产生负面的影响。但有时可能需要与那些使用阻塞 API 的遗留代码进行交互。对于这种情况，ChannelPipeline 有一些接受一个 EventExecutorGroup 的 add()方法。如果一个事件被传递给一个自定义的 EventExecutorGroup，它将被包含在这个 EventExecutorGroup 中的某个 EventExecutor 所处理，从而被从该 Channel 本身的 EventLoop 中移除。对于这种用例，Netty 提供了一个叫 DefaultEventExecutorGroup 的默认实现。

ChannelPipeline的用于访问ChannelHandler的操作：