Netty 系列四（ChannelHandler 和 ChannelPipeline）.

2021-09-08 11:50:58

一、概念

先来整体的介绍一下这篇博文要介绍的几个概念（Channel、ChannelHandler、ChannelPipeline、ChannelHandlerContext、ChannelPromise）：

Channel：Netty 中传入或传出数据的载体；
ChannelHandler：Netty 中处理入站和出站数据的应用程序逻辑的容器；
ChannelPipeline：ChannelHandler链的容器；
ChannelHandlerContext：代表了 ChannelHandler 和 ChannelPipeline 之间的关联，每当有ChannelHandler 添加到 ChannelPipeline 中时，都会创建 ChannelHandlerContext；
ChannelPromise：ChannelPromise是ChannelFuture的一个子类，其定义了一些可写的方法，如setSuccess()和setFailure()，从而使ChannelFuture不可变。

我们来举一个例子描述这些概念之间的逻辑关系：服务端接收到客户端的连接请求，创建一个Channel同客户端进行绑定，新创建的 Channel 会都将会被分配一个新的ChannelPipeline（这项关联是永久性的，Channel 既不会附加另外一个ChannelPipeline，也不能分离当前的）。而 ChannelPipeline 作为 ChannelHandler链的容器，当Channel 生命周期中状态发生改变时，将会生成对应的事件，这些事件将会被 ChannelPipeline 中 ChannelHandler 所响应，响应方法的参数一般都有一个 ChannelHandlerContext ，一个 ChannelHandler 对应一个 ChannelHandlerContext，ChannelHandlerContext 代表了 ChannelHandler 和 ChannelPipeline 之间的关联，这项关联也是永远不会改变的。

二、ChannelHandler

Netty提供了大量预定义的可以开箱即用的ChannelHandler实现，包括用于各种协议的ChannelHandler。因此，我们在自定义ChannelHandler实现用于处理我们的程序逻辑时，只需要继承Netty 的一些默认实现即可，主要有两种：

1、继承 ChannelHandlerAdapter （在4.0 中处理入站事件继承 ChannelInboundHandlerAdapter，处理出站事件继承 ChannelOutboundHandlerAdapter ；在5.0 推荐直接继承 ChannelHandlerAdapter）

2、继承 SimpleChannelInboundHandler

这两种方式有什么区别呢？当我们处理入站数据和出站数据时，都需要确保没有任何的资源泄露。在入站方向，继承 SimpleChannelInboundHandler 的实现类会在消息被处理之后自动处理消息，而继承 ChannelHandlerAdapter 的实现类需要手动的释放消息（ReferenceCountUtil.release(msg)）；在出站方向，不管继承的是哪一种的实现类，当你处理了 write() 操作并丢弃了一个消息，那么你就应该释放它，不仅如此，还要通知 ChannelPromise。否则可能会出现 ChannelFutureListener 收不到某个消息已经被处理了的通知的情况。

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ReferenceCountUtil.release(msg);
    //ChannelPromise 是ChannelFuture的一个子类，设置成 true 通知 ChannelFutureListener 消息已经被处理了
    //当一个 Promise 被完成之后，其对应的 Future 的值便不能再进行任何修改了
    promise.setSuccess();
}

tips:总之，如果一个消息被消费或者丢弃了，并且没有传递给 ChannelPipeline 中的下一个ChannelOutboundHandler，那么用户就有责任调用 ReferenceCountUtil.release()。

来看看 ChannelHandler 的 API：

isSharable ：如果其对应的实现被标注为 Sharable，那么这个方法将返回 true，表示它可以被添加到多个 ChannelPipeline中

-- ChannelHandler 生命周期方法 --
handlerAdded ：当把 ChannelHandler 添加到 ChannelPipeline 中时被调用
handlerRemoved ：当从 ChannelPipeline 中移除 ChannelHandler 时被调用
exceptionCaught ：当处理过程中在 ChannelPipeline 中有错误产生时被调用

-- 处理入站数据以及各种状态变化 --
channelRegistered : 当 Channel 已经注册到它的 EventLoop 并且能够处理 I/O 时被调用
channelUnregistered : 当 Channel 从它的 EventLoop 注销并且无法处理任何 I/O 时被调用
channelActive : 当 Channel 处于活动状态时被调用；Channel 已经连接/绑定并且已经就绪
channelInactive : 当 Channel 离开活动状态并且不再连接它的远程节点时被调用
channelReadComplete : 当Channel上的一个读操作完成时被调用
channelRead : 当从 Channel 读取数据时被调用
ChannelWritabilityChanged :当 Channel 的可写状态发生改变时被调用。
userEventTriggered : 当 ChannelnboundHandler.fireUserEventTriggered()方法被调用时被调用，因为一个 POJO 被传经了 ChannelPipeline

-- 处理出站数据并且允许拦截所有的操作 --
bind : 当请求将 Channel 绑定到本地地址时被调用
connect : 当请求将 Channel 连接到远程节点时被调用
disconnect : 当请求将 Channel 从远程节点断开时被调用
close : 当请求关闭 Channel 时被调用
deregister : 当请求将 Channel 从它的 EventLoop 注销时被调用
read : 当请求从 Channel 读取更多的数据时被调用
flush : 当请求通过 Channel 将入队数据冲刷到远程节点时被调用
write :当请求通过 Channel 将数据写到远程节点时被调用

三、ChannelPipeline

ChannelPipeline 是一个拦截流经 Channel 的入站和出站事件的ChannelHandler 实例链，它和 ChannelHandler 之间的交互组成了应用程序数据和事件处理逻辑的核心，而它们之间的关联交互就是通过 ChannelHandlerContext。

如果一个入站事件被触发，它将被从 ChannelPipeline 的头部开始一直被传播到 Channel Pipeline 的尾端。如图，Netty 总是将 ChannelPipeline 的入站口作为头部，而将出站口作为尾端，如图，第一个被入站事件看到的 ChannelHandler 将是1，而第一个被出站事件看到的是 ChannelHandler 将是 5。

既然 ChannelPipeline 是 ChannelHandler链的容器，让我们来看看ChannelPipeline 是如何管理 ChannelHandler的吧！

addFirst : 将一个 ChannelHandler 添加到 ChannelPipeline 最开始位置中
addBefore ：将一个 ChannelHandler 添加到 ChannelPipeline 某个ChannelHandler前
addAfter：将一个 ChannelHandler 添加到 ChannelPipeline 某个ChannelHandler后
addLast : 将一个 ChannelHandler 添加到 ChannelPipeline 最末尾位置
remove ：将一个 ChannelHandler 从 ChannelPipeline 中移除
replace ：将 ChannelPipeline 中的一个 ChannelHandler 替换为另一个 ChannelHandler
get ：通过类型或者名称返回 ChannelHandler
context ：返回和 ChannelHandler 绑定的 ChannelHandlerContext
names ：返回 ChannelPipeline 中所有 ChannelHandler 的名称

ChannelPipeline 的API 用于调用入站操作的附加方法：

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, Object msg)方法
fireChannelReadComplete：调用 ChannelPipeline 中下一个 ChannelInboundHandler 的channelReadComplete(ChannelHandlerContext)方法
fireChannelWritabilityChanged：调用 ChannelPipeline 中下一个 ChannelInboundHandler 的channelWritabilityChanged(ChannelHandlerContext)方法

ChannelPipeline 的API 用于调用出站操作的附加方法：

bind：将 Channel 绑定到一个本地地址，这将调用 ChannelPipeline 中的下一个ChannelOutboundHandler 的 bind方法
connect：将 Channel 连接到一个远程地址，这将调用 ChannelPipeline 中的下一个ChannelOutboundHandler 的 connect方法
disconnect：将 Channel 断开连接。这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 disconnect方法
close：将 Channel 关闭。这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 close方法
deregister：将 Channel 从它先前所分配的 EventExecutor（即 EventLoop）中注销。这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler 的 deregister方法
flush：冲刷Channel所有挂起的写入。这将调用ChannelPipeline中的下一个ChannelOutboundHandler 的 flush方法
write：将消息写入 Channel。这将调用 ChannelPipeline 中的下一个 ChannelOutboundHandler的write方法。注意：这并不会将消息写入底层的 Socket，而只会将它放入队列中。要将它写入 Socket，需要调用 flush()或者 writeAndFlush()方法
writeAndFlush：这是一个先调用 write()方法再接着调用 flush()方法的便利方法
read：请求从 Channel 中读取更多的数据。这将调用 ChannelPipeline 中的下一个ChannelOutboundHandler 的 read(ChannelHandlerContext)方法

四、ChannelHandlerContext

ChannelHandlerContext 代表了 ChannelHandler 和 ChannelPipeline 之间的关联。

ChannelHandlerContext 有很多的方法，其中一些方法也存在于 Channel 和 ChannelPipeline 本身上，但是有一点重要的不同。如果调用 Channel 或者 ChannelPipeline 上的这些方法，它们将沿着整个 ChannelPipeline 进行传播。而调用位于 ChannelHandlerContext上的相同方法，则将从当前所关联的 ChannelHandler 开始，并且只会传播给位于该ChannelPipeline 中的下一个能够处理该事件的 ChannelHandler。因此，尽量使用 ChannelHandlerContext 的同名方法来处理逻辑，因为它将产生更短的事件流，应该尽可能地利用这个特性来获得最大的性能。

五、异常处理

入站异常处理：

1、ChannelHandler.exceptionCaught()的默认实现是简单地将当前异常转发给ChannelPipeline 中的下一个 ChannelHandler；
2、如果异常到达了 ChannelPipeline 的尾端，它将会被记录为未被处理；
3、要想定义自定义的处理逻辑，你需要重写 exceptionCaught()方法。一般将这个Channelhandler放在 ChannelPipeline 的最后，确保所有的入站异常都总会被处理。

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    cause.printStackTrace();
    ctx.channel().close();
}

出站异常处理：

@Override
public void channelActive(ChannelHandlerContext ctx) {
    ChannelFuture channelFuture = ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rock!", CharsetUtil.UTF_8));
    //出站异常处理
    channelFuture.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                future.cause().printStackTrace();
                future.channel().close();
            }
        }
    });
}

六、寄语

有时候也会迷茫，身边的人理论基础差一些的的，代码不一样敲的好好的？而我花大量时间细细的去研究这么理论真的值得吗？仔细想想，人生很多事情本来就是徒劳无功的啊，没必要急功近利，欲速则不达。要坚信，一切的付出总会在人生的某个时刻回报在我们身上。

码农公寓