Netty 框架学习 —— ChannelHandler 与 ChannelPipeline
ChannelHandler
1. Channel 生命周期
Channel 的生命周期状态如下：
| 状态 | 描述 |
|---|---|
| ChannelUnregistered | Channel 已经被创建，但还未注册到 EventLoop |
| ChannelRegistered | Channel 已经被注册到 EventLoop |
| ChannelActive | Channel 处于活动状态（已经连接到它的远程节点），可以接收和发送数据 |
| ChannelInactive | Channel 没有连接到远程节点 |
Channel 的生命周期按上表从上到下所示，当状态发生改变时，将会生成对应的事件。这些事件将会被转发到 ChannelPipeline 中的 ChannelHandler，随后其可以做出响应
2. ChannelHandler 生命周期
下表列出了 ChannelHandler 定义的生命周期操作，在 ChannelHandler 被添加到 ChannelPipeline 或者被从 ChannelPipeline 移除时会调用这些操作，这些方法中的每一个都接受一个 ChannelHandlerContext 参数
| 类型 | 描述 |
|---|---|
| handlerAdded | 当把 ChannelHandler 添加到 ChannelPipeline 中时被调用 |
| handlerRemoved | 当从 ChannelPipeline 中移除 ChannelHandler 时被调用 |
| exceptionCaught | 当处理过程中在 ChannelPipeline 中有错误产生时被调用 |
3. ChannelInboundHandler 接口
ChannelInboundHandler 是 ChannelHandler 接口的子接口，处理入站数据以及各种状态变化。下表列出 ChannelInboundHandler 的生命周期方法，这些方法将会在数据被接收时或者与其对应的 Channel 状态发生改变时被调用
| 类型 | 描述 |
|---|---|
| channelRegistered | 当 Channel 已经注册到它的 EventLoop 并且能够处理 IO 时被调用 |
| channelUnregistered | 当 Channel 从它的 EventLoop 注销并且无法处理任何 IO 时被调用 |
| channelActive | 当 Channel 处于活动状态时被调用，Channel 已经连接/绑定并且就绪 |
| channelInactive | 当 Channel 离开活动状态并且不再连接它的远程节点时被调用 |
| channelReadComplete | 当 Channel 上的一个读操作完成时被调用 |
| channelRead | 当从 Channel 读取数据时被调用 |
| ChannelWritabilityChanged | 当 Channel 的可写状态发生改变时被调用 |
| useEventTriggered | 当 ChannelInboundHandler.fireUserEventTriggered() 方法被调用时被调用，因为一个 POJO 流经 ChannelPipeline |
当某个 ChannelInboundHandler 的实现重写 channelRead() 方法时，它将负责显式地释放与池化 ByteBuf 实例相关的内存。Netty 为此提供了一个实用方法 ReferenceCountUtil.release()
@Sharable
public class DiscardHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ReferenceCountUtil.release(msg);
}
}
这种方式可能会很烦琐，另一种更加简单的方式是使用 SimpleChannelInboundHandler
@Sharable
public class SimpleDiscardHandler extends SimpleChannelInboundHandlerAdapter<Object> {
@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) {
// 不需要显式进行资源释放
}
}
由于 SimpleChannelInboundHandler 会自动释放资源，所以你不应该存储任何指向消息的引用，因为这些引用都将会失效
4. ChannelOutboundHandler 接口
出站操作和数据由 ChannelOutboundHandler 处理，它的方法将被 Channel、ChannelPipeline 以及 ChannelHandlerContext 调用
ChannelOutboundHandler 的一个强大的功能是可以按需推迟操作或者事件，使得可以通过一些复杂的方法来处理请求，例如，如果到远程节点的写入被暂停了，那么你可以推迟冲刷操作并在稍后继续
下标显示了所有由 ChannelOutboundHandler 本身所定义的方法
| 类型 | 描述 |
|---|---|
| bind(ChannelHandlerContext, SocketAddress, ChannelPromise) | 当请求将 Channel 绑定到本地地址时被调用 |
| connect(ChannelHandlerContext, SocketAddress, ChannelPromise) | 当请求将 Channel 连接到远程节点时被调用 |
| disconnect(ChannelHandlerContext, ChannelPromise) | 当请求将 Channel 从远程节点断开时被调用 |
| close(ChannelHandlerContext, ChannelPromise) | 当请求关闭 Channel 时被调用 |
| deregister(ChannelHandlerContext, ChannelPromise) | 当请求将 Channel 从它的 EventLoop 注销时被调用 |
| read(ChannelHandlerContext) | 当请求从 Channel 读取更多的数据时被调用 |
| flush(ChannelHandlerContext) | 当请求通过 Channel 将入队数据冲刷到远程节点时被调用 |
| write(ChannelHandlerContext, Object, ChannelPromise) | 当请求通过 Channel 将数据写到远程节点时被调用 |
ChannelOutboundHandler 中的大部分方法都需要一个 ChannelPromise 参数，以便在操作完成时得到通知。CHannelPromise 是 ChannelFuture 的一个子类，定义了一些可写方法
5. ChannelHandler 适配器
可以使用 ChannelInboundHandlerAdapter 和 ChannelOutboundHandlerAdapter 类作为自己的 Channel 的起始点，这两个适配器分别提供了 ChannelInboundHandler 和 CHannelOutboundHandler 的基本实现，并扩展抽象类 ChannelHandlerAdapter
ChannelHandlerAdapter 还提供了实用方法 isSharable()，如果其对应的实现被标记为 Sharable，那么这个方法将返回 true，表示它可以被添加到多个 ChannelPipeline 中
如果想在自己的 ChannelHandler 中使用这些适配器类，只需要简单地扩展它们，并重写自定义方法
ChannelPipeline 接口
可以把 ChannelPipeline 理解成一个拦截流经 Channel 的入站和出站事件的 ChannelHandler 实例链，每一个新创建的 Channel 都将会被分配一个新的 ChannelPipeline，Channel 不能附加到另一个 ChannelPipeline，也不能从当前分离。根据事件的起源，事件将会被 ChannelInboundHandler 或者 ChannelOutboundHandler 处理，随后，通过调用 ChannelHandlerContext 实现，它将被转发给同一超类型的下一个 ChannelHandler
一个典型的同时具有入站和出站 ChannelHandler 的 ChannelPipeline 布局如图所示
在 ChannelPipeline 传播事件时，它会测试 ChannelPipeline 中的下一个 ChannelHandler 的类型是否和事件的运动方向相匹配，如果不匹配，ChannelPipeline 将跳过该 ChannelHandler 并前进到下一个，直到找到和该事件所期望的方向相匹配
1. 修改 ChannelPipeline
通过调用 ChannelPipeline 上的相关方法，ChannelHandler 可以添加、删除或者替换其他的 ChannelHandler，当然也包括它自己
下表列出了由 ChannelHandler 修改 ChannelPipeline 的相关方法
| 名称 | 描述 |
|---|---|
| addFirst addBefore addAfter addLast |
将一个 ChannelHandler 添加到 ChannelPipeline |
| remove | 将一个 ChannelHandler 从 ChannelPipeline 中移除 |
| replace | 将 ChannelPipeline 中的一个 ChannelHandler 替换为另一个 ChannelHandler |
ChannelPipeline 的用于访问 ChannelHandler 的操作
| 名称 | 描述 |
|---|---|
| get | 通过类型或者名称返回 ChannelHandler |
| context | 返回和 ChannelHandler 绑定的 ChannelHandlerContext |
| names | 返回 ChannelPipeline 中所有 ChannelHandler 的名称 |
2. 触发事件
ChannelPipeline 的 API 公开了用于调用入站和出站操作的附加方法
ChannelPipeline 的入站操作如表
| 方法名称 | 描述 |
|---|---|
| fireChannelRegistered | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelRegistered(ChannelHandlerContext) 方法 |
| fireChannelUnregistered | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelUnregistered(ChannelHandlerContext) 方法 |
| fireChannelActive | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelActive(ChannelHandlerContext) 方法 |
| fireChannelInactive | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelInactive(ChannelHandlerContext) 方法 |
| fireExceptionCaught | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 exceptionCaught(ChannelHandlerContext, Throwable) 方法 |
| fireUserEventTriggered | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 userEventTriggered(ChannelHandlerContext, Object) 方法 |
| fireChannelRead | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelRead(ChannelHandlerContext) 方法 |
| fireChannelReadComplete | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelReadComplete(ChannelHandlerContext) 方法 |
| fireChannelWritabilityChanged | 调用 ChannelPipeline 中下一个 ChannelInboundHandler 的 channelWritabilityChanged(ChannelHandlerContext) 方法 |
ChannelPipeline 的出站操作如表
| 方法名称 | 描述 |
|---|---|
| bind | 将 Channel 绑定到一个本地地址，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 bind(ChannelHandlerContext, SocketAddress, ChannelPromise) 方法 |
| connect | 将 Channel 绑定到一个远程地址，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 connect(ChannelHandlerContext, SocketAddress, ChannelPromise) 方法 |
| disconnect | 将 Channel 断开连接，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 disconnect(ChannelHandlerContext, ChannelPromise) 方法 |
| close | 将 Channel 关闭，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 close(ChannelHandlerContext, ChannelPromise) 方法 |
| deregister | 将 Channel 从它先前所分配的 EventExecutor（即 EventLoop）中注销，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 deregister(ChannelHandlerContext, ChannelPromise) 方法 |
| flush | 冲刷 Channel 所有挂起的写入，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 flush(ChannelHandlerContext) 方法 |
| write | 将消息写入 Channel，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 write(ChannelHandlerContext, Object, ChannelPromise) 方法 |
| writeAndFlush | 这是一个先调用 write() 方法再接着调用 flush() 方法的便利方法 |
| read | 请求从 Channel 中读取更多的数据，这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 read(ChannelHandlerContext) 方法 |
总结一下：
- ChannelPipeline 保存了与 Channel 相关联的 ChannelHandler
- ChannelPipeline 可以根据需要，动态修改 ChannelHandler
- ChannelPipeline 有丰富的 API 用以响应入站和出站事件
ChannelContext 接口
ChannelHandlerContext 代表了 ChannelHandler 和 ChannelPipeline 之间的关联，每当有 ChannelHandler 添加到 ChannelPipeline 中时，都会创建 ChannelHandlerContext。ChannelHandlerContext 的主要功能是管理它所关联的 ChannelHandler 和在同一个 ChannelPipeline 中的其他 ChannelHandler 之间的交互
ChannelHandlerContext 有很多方法，其中一些方法也存在于 Channel 和 ChannelPipeline 上，不同的是，如果调用 Channel 或者 ChannelHandlerPipeline 上的这些方法，它们将沿着整个 ChannelPipeline 进行传播。而调用位于 ChannelHandlerContext 上的相同方法，则将从当前所关联的 ChannelHandler 开始，并且只会传播给位于该 ChannelPipeline 中的下一个能处理该事件的 ChannelHandler
1. 使用 ChannelHandlerContext
Channel、ChannelPipeline、ChannelHandler 以及 ChannelHandlerContext 之间的关系如图所示
通过 ChannelHandler 或 ChannelPipeline 操作，事件将传播整个 ChannelPipeline，但在 ChannelHandler 的级别上，事件从上一个 ChannelHandler 到下一个 ChannelHandler 的移动是由 ChannelHandlerContext 上的调用完成的
为什么想要从 ChannelPipeline 中的某个特定点开始传播事件？
- 减少将事件传经对它不感兴趣的 ChannelHandler 所带来的开销
- 避免将事件传经那些可能会对它感兴趣的 ChannelHandler
要想调用从某个特定的 ChannelHandler 开始的处理过程，必须获取该 ChannelHandler 之前的 ChannelHandler 所关联的 ChannelHandlerContext，这个 ChannelHandlerContext 将调用和它所关联的 ChannelHandler 之后的 ChannelHandler
我们还可以通过调用 ChannelHandlerContext 上的 pipeline() 方法来获得 ChannelPipeline 的引用，进而得以在运行时操作 ChannelHandler
异常处理
1. 处理入站异常
如果在处理入站事件的过程中有异常抛出，那么它将从它在 ChannelInboundHandler 里被触发的那一点开始流经 ChannelPipeline。要想处理这种类型的入站异常，你需要在你的 ChannelInboundHandler 实现重写以下方法，否则默认将异常转发给下一个 ChannelHandler：
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
因为异常会按照入站方向继续流动，因此实现了处理异常逻辑的 ChannelInboundHandler 通常位于 ChannelPipeline 的最后，确保异常总能被处理
2. 处理出站异常
用于处理出站操作的正常完成以及异常的选项，都基于以下通知机制：
- 每个出站操作都将返回一个 ChannelFuture，注册到 ChannelFuture 的 ChannelFutureListener 将在操作完成时通知结果
- 几乎所有的 ChannelOutboundHandler 上的方法都会传入一个 ChannelPromise 实例，作为 ChannelFuture 的子类，ChannelPromise 也可以被分配用于异步通知的监听器
添加 ChannelFutureListener 只需要调用 ChannelFuture实例上的 addListener(ChannelFutureListener) 方法，并且有两种方式可以做到，第一种是调用出站操作（如 write 方法）所返回的 ChannelFuture 上的 addListener() 方法；第二种是 ChannelFutureListener 添加到作为参数传递的 ChannelOutboundHandler 的方法的 ChannelPromise
所以，如果你的 ChannelOutboundHandler 抛出异常，Netty 本身会通知任何已经注册到对应 ChannelPromise 的注册器