Netty in Action ——— ChannelHandl

2017-10-30 本文已影响543人 tomas家的小拨浪鼓

本文是Netty文集中“Netty in action”系列的文章。主要是对Norman Maurer and Marvin Allen Wolfthal 的《Netty in action》一书简要翻译，同时对重要点加上一些自己补充和扩展。

概要

ChannelHandler 和 ChannelPipeline 的 API
资源泄漏检测
异常处理

The ChannelHandler family

Channel生命周期

Channel接口定义了简单但强大的状态模式来紧密的联系ChannelInboundHandler API。包括了4种状态

Channel正常生命周期状态改变如下图：

当遇到状态改变时，相应的事件会被产生。该事件会在ChannelPipeline中的ChannelHandlers传递。

ChannelHandler的生命周期

当一个ChannelHandler添加到一个ChannelPipeline或从一个ChannelPipeline中移除时，会调用ChannelHandler生命周期中相应的方法。每个方法都会接收一个ChannelHandlerContext参数。

注意，Channel生命周期状态改变时会有相应的事件产生，并且该事件会在ChannelPipeline中的ChannelHandlers传递。但，ChannelHandler生命周期状态改变时并不会有相应的事件产生与ChannelPipeline中传播，只有回调当前这个ChannelHandler的某个方法而已。这是很好理解的，因为ChannelPipeline中所有的ChannelHandler都是服务于一个Channel的，因此Channel的状态改变了，ChannelPipeline中的ChannelHandler都有可能需要做出相应的处理。而ChannelHandler自身的状态和其他ChannelHandler并无关系，所以ChannelHandler状态发生改变时，会有当前这个ChannelHandler相应的方法会被回调而已。

Netty定义了两个非常重要的ChannelHandler的子接口：

ChannelInboundHandler —— 处理入站数据和所有类型状态的改变
ChannelOutboundHandler —— 处理所有的出站数据和允许拦截所有的操作。

ChannelInboundHandler 接口

下表列出了ChannelInboundHandler接口生命周期的所有方法。这些方法将被调用在接收数据或者Channel相关状态改变时。正如我们早前提及的，这些方法与Channel的生命周期紧密映射。

当一个ChannelInboundHandler的实现重写了channelRead()，它需要负责明确释放ByteBuf实例相关的内存。为了达到这个目的，Netty提供了一个实用的方法：ReferenceCountUtil.release()

Netty会通过警告级别的日志消息打印未释放的资源，这将能够简单的从代码中寻找出未释放的实例。但是通过这种方式管理资源是非常笨拙的。一个简单的替换方式是使用SimpleChannelInboundHandler，如下所示

因为SimpleChannelInboundHandler会自动释放资源，你不能够存储任何消息的引用用于后面再次使用，因为这些消息会变得无效。

ChannelOutboundHandler 接口

通过ChannelOutboundHandler进行出站的操作和数据的处理。ChannelOutboundHandler的方法可以通过Channel、ChannelPipeline、ChannelHandlerContext调用。
ChannelOutboundHandler有一个强大的功能是根据需要可延迟一个操作或事件，这将允许通过一个复杂的方式去处理请求。例如，如果写数据到远端被暂停了，你能够延迟刷新操作并稍后恢复写操作。

ChannelPromise VS ChannelFuture
ChannelOutboundHandler的许多方法都会接受一个ChannelPromise参数用于当操作完成时得到一个通知。ChannelPromise是ChannelFuture的一个子类，ChannelPromise定义了可写入的方法，比如：setSuccess()或setFailure()，它使得ChannelFuture不可变( 因为，方法参数使用final域修饰，如下例子所示 )。
也就是说，ChannelPromise是一个特殊的ChannelFuture，ChannelPromise是可写的。从源码上看，ChannelOutboundHandler的write方法返回的ChannelFuture实际上就是ChannelPromise，ChannelPromise会在ChannelPipeline中传递。

ChannelHandler 适配器

你能够使用ChannelInboundHandlerAdapter类和ChannelOutboundHandlerAdapter类作为你自己ChannelHandlers的入口。这两个适配器分别提供了ChannelInboundHandler和ChannelOutboundHandler的基本实现。

ChannelHandlerAdapter提供了一个非常实用的方法isSharable()。如果ChannelHandlerAdapter的实现类被注解了"@Sharable”，那么该方法将返回true，表明着该实现类实例能被添加到多个ChannelPipeline中，或者可以被添加到同个ChannelPipeline中多次。
ChannelInboundHandlerAdapter和ChannelOutboundHandlerAdapter中方法通过调用ChannelHandlerContext等价的方法来将事件传递给pipeline中的下一个ChannelHandler。如：

你自己的处理器可以简单的实现适配器类，然后重写你想要自定义的方法。

资源管理

无论你通过调用 ChannelInboundHandler.channelRead() 或 ChannelOutboundHandler.writer() 来处理你的数据，你需要确保没有资源的泄漏。
Netty使用引用计数来处理ByteBufs。所以在你使用完一个ByteBuf时去调整引用计数是非常重要的。
为了帮助你诊断可能存在的问题，Netty提供了ResourceLeakDetector类，它将获取你应用已经分配的缓存中的1%数据作为采样来进行内存泄漏的检查。它的开销是非常小的。

如果一个泄漏被检测出来了，日志将打印类似下面的消息：

泄漏检查等级的设置是通过在下面的java系统属性来设置的：java -Dio.netty.leakDetectionLevel=ADVANCED或者通过调用ResourceLeakDetector.setLevel()方法来设置。

下面是一个典型的单元测试产生的泄漏报告：

当入站消息不再传递给下一个ChannelInboundHandler时，通过ReferenceCountUtil.release(msg);来释放资源。

消费入站消息的简单方式：Netty提供了一个叫做SimpleChannelInboundHandler的ChannelInboundHandler的实现，该实现将自动释放一个消息，当该消息被channelRead0()消费后。源码如下：

对于出站来说，如果你处理了一个write()操作并且需要废弃一个消息，那么你需要负责去释放该消息。

注意，这里不仅需要释放资源，还需要通知ChannelPromise(即，上面的例子是通过调用promise.setSuccess()来通知异步的write操作已经成功完成)。
另一方面，还一种情况可能出现：ChannelFutureListener可能没有收到关于一个消息已经被处理的通知。

总而言之，用户有责任去通过调用ReferenceCountUtil.release()来释放一个已经被消费的消息或废弃并不会传递给ChannelPipeline中下一个ChannelOutboundHandler的消息。如果消息到真实的传输层，当他写完或Channel被关闭时将会被自动释放。

重要：

内存泄漏针对于使用了池的ByteBuf，在从池中分配完ByteBuf后使用完又没有放回到池中。如，从池中分配ByteBuf，ctx.alloc()默认返回PooledByteBufAllocator：
👆这里要说明一点，如果是在正式项目中使用，没有rep.copy()，而是直接将rep传给writeAndFlush(…)是不会导致内存泄漏的，因为出站操作时，编码器encoder调用会自动释放资源。如果我们在正式项目中写rep.release();反而会导致错误。
而在使用EmbeddedChannel测试入站操作时，直接将rep传给writeAndFlush(…)也是可以测出内存泄漏的，因为EmbeddedChannel测试入站操作时没有走出站流程，所以就导致从池中分配的ByteBuf一直无法得到释放。
内存泄漏报告： 非池分配，则不会有内存泄漏报告：
writeAndFlush(Bytebuf buf)，如果是EmbeddedChannel单元测试且测试入站数据时，需要手动释放资源
如果是EmbeddedChannel的出站数据测试，或正常程序运行，则不需要。因为编码器的encode()方法调用会自动释放资源。如果再手动调用反倒会报错：

Q：所以如果是涉及到I/O操作的就不需要用户去释放出站资源了吗？因为数据在网络层传输出去后会被自动释放？
A：展开下encoder的底层：
MessageToMessageEncoder：如果有执行encode(…)函数，则会在finally中释放资源，因为调用了encode函数则会生成新的ByteBuf或其他对象传递给下一个Encoder，那么当前传入的msg就已经不会再使用了，所以需要释放msg。
如果没有执行encode(…)函数，则直接将msg传给下一个Encoder，由下一个Encoder执行一样的操作。

MessageToByteEncoder：也会有相关的释放资源的操作。

ChannelPipeline 接口

每一个新的Channel被创建时都会分配一个新的ChannelPipeline。他们的关系是永久不变的，Channel既不能依附于其他ChannelPipeline也不能和当前ChannelPipeline分离。这是一个固定的操作在Netty组件的生命周期中并且不需要开发者做任何的额外的操作。
一个事件要么被一个ChannelInboundHandler处理要么被一个ChannelOutboundHandler处理。随后该事件通过一个ChannelHandlerContext来实现传递给下一个具有一样父类的处理器，即同一个方向的处理器。

ChannelHandlerContext
ChannelHandlerContext使一个ChannelHandler能和它所在的ChannelPipeline中的其他的ChannelHandler交互。一个处理器能够通知ChannelPipeline中下一个ChannelHandler并且能够动态修改它所属的ChannelPipeline。
也就是说，事件是通过ChannelHandlerContext在ChannelPipeline中的ChannelHandler传递的。处理器在获得ChannelPipeline后，可以对其进行动态修改，即修改后即生效。

ChannelPipeline根本上是一系列的ChannelHandlers。ChannelPipeline还提供了方法用于传播事件通过ChannelPipeline本身。

ChannelHandler在ChannelPipeline中的顺序是由我们通过ChannelPipeline.add*()方法来确定的。

当管道传播一个事件时，它会确定是否管道中下一个ChannelHandler符合移动的方向。如果不符合，ChannelPipeline会跳过这个ChannelHandler并继续操作下一个，直到它找到一个符合它想要的方向的ChannelHandler。(当然，一个handler可能同时实现了ChannelInboundHandler和ChannelOutboundHandler接口)

修改一个ChannelPipeline

ChannelHandler能够实时修改ChannelPipeline的布局，通过添加、删除、替换为其他ChannelHandlers ( 它也能将自己从ChannelPipeline中移除 )。这是ChannelHandler非常重要的功能。

ChannelHandler 的执行和阻塞
通常，ChannelPipeline中的每个ChannelHandler将事件传递给它所在的EventLoop ( I/O线程 )进行处理。非常重要的是不要阻塞这个线程，这样做会对该线程上所有的I/O处理造成负面影响。
有时候我们可能需要对接使用了阻塞的遗留代码。在这种情况下，可以通过ChannelPipeline的add()方法来接收一个EventExecutorGroup参数。如果一个事件被传给了一个自定义的EventExecutorGroup，它将被这个EventExecutorGroup中的某个EventExecutor所处理，并且会从Channel所在的EventLoop中移除。对于这种使用情况Netty提供了一个叫做DefaultEventExecutorGroup的实现。

也就是说，只有传入的EventExecutorGroup的ChannelHandler才会放到EventExecutorGroup中执行，不会影响到ChannelPipeline中其他的ChannelHandler。
在通过ChannelHandlerContext将事件传播给下一个ChannelHandler的时候，又会依据下一个ChannelHandler关联的ChannelHandlerContext的executor()获得执行ChannelHandler的执行器，如果在将ChannelHandler添加到ChannelPipeline中时没有指定执行的EventExecutorGroup，那么默认就是在Channel所注册到的EventLoop所在线程上执行。

事件触发

ChannelPipeline API 暴露了附加的方法用于调用入站和出站的操作。

在出站方面，处理一个事件将导致底层socket采取某些动作。

注意，ChannelPipeline附加的对入站和出站的操作，都只是触发ChannelPipeline中消息从管道头(入站操作)或管道尾(出站操作)开始处理该消息，ChannelPipeline这些方法本身并不会去对事件做一个逻辑处理。如下：

总结：

一个ChannelPipeline持有一个Channel相关的所有ChannelHandlers。
能够根据需要通过添加和删除来动态修改一个ChannelPipeline。
ChannelPipeline拥有丰富的API用于调用action来回应入站和出站事件。

ChannelHandlerContext 接口

一个ChannelHandlerContext代表一个ChannelHandler和一个ChannelPipeline之间的关联，并且无论何时一个ChannelHandler被添加到一个ChannelPipeline时它都会被创建。ChannelHandlerContext一个主要的功能是管理与它相关的ChannelHandler在ChannelPipeline中与其他ChannelHandler间的交互。
ChannelHandlerContext有许多的方法，有些方法也出现在了Channel和ChannelPipeline类中，但是它们有着很重要的不同处。如果你通过一个Channel实例或ChannelPipeline实例调用这些方法，它们将会传播通过整个管道。但相同的方法通过ChannelHandlerContext被调用时，它将从当前关联的ChannelHandler开始并传播给管道中下一个能够处理该事件的ChannelHandler。
也就是说，ChannelPipeline和Channel中调用的方法都会通过整个管道；而ChannelHandlerContext调用的方法会从当前ChannelHandler对应方向的下一个ChannelHandler开始执行( 当前ChannelHandler也不会处理该事件 )。

从源码上看write操作是当前ChannelHandler对应的下一个ChannelOutboundHandler开始处理写操作，注意，出站的方向是从tail -> head;

Q：从👆的API来看，read方法是会从pipeline中第一个入站缓冲区，read是通过整个pipeline吗？
A：是的，Channel会将每次读取到的数据传到pipeline中。注意，这里是每一次读到的数据，而不是读到完整的消息或全部读完数据，所以才有后面需要handler来解析收到的数组以组装成一个消息。
ChannelPipeline中的head的invokeChannelRead(..)的操作就是执行AbstractChannelHandlerContext的fireChannelRead方法，将msg传递给pipeline中的下个ChannelInboundHandler。

当使用ChannelHandlerContext API时请注意下面几点：

ChannelHandlerContext关联的ChannelHandler是不会改变的，所以缓存ChannelHandlerContext引用是安全的。
与其他类中有着相同名字的有效方法比较，ChannelHandlerContext方法提供了一个更短的事件流。这可被利用于提供一个更好的性能。

ChannelHandlerContext 的使用

从一个ChannelHandlerContext对象来访问Channel

从一个ChannelHandlerContext对象来访问ChannelPipeline

上面两个例子是相同的。这里重要的是要注意，当通过Channel或ChannelPipeline调用write()的时候，事件会传播通过整个管道。它们通过ChannelHandlerContext将事件从一个处理器传递到下一个处理器。

为什么你想要传播一个事件从ChannelPipeline中的指定的某一点开始？

减小传递事件通过不感兴趣的ChannelHandlers的开销。即，通过事件不传递给不感兴趣的ChannelHandler来减小开销。
为了防止对事件感兴趣的处理器处理事件

为了调用程序从一个指定的ChannelHandler开始，你必须引用一个指定ChannelHandler前一个的ChannelHandler相关联的ChannelHandlerContext。
👆这里再次说明，通过ChannelHandlerContext调用方法时，当前ChannelHandler是不会对事件进行处理的，而是从下一个ChannelHandler开始对事件进行处理。

ChannelHandler 和 ChannelHandlerContext 的高级用法

你能够通过ChannelHandlerContext的pipeline()方法获取的一个ChannelPipeline引用。它能在运行时操作管道中的ChannelHandlers，这能够被利用与实现复杂的设计。比如，你能够添加一个ChannelHandler到一个管道以支持一个动态协议的改变。
另一个高级用法是支持缓存一个ChannelHandlerContext引用，用于稍后使用，该引用能在任何ChannelHandler以外的方法中使用，甚至可以在不同的线程中使用。

因为一个ChannelHandler能够属于多个ChannelPipeline，它能够被绑定到多个ChannelHandlerContext实例上。一个用于该目的的ChannelHandler必须有@Sharable注解；如果一个不具有@Sharable注解的ChannelHandler被尝试去添加到多个ChannelPipeline中将会触发一个异常。显然，为了安全的在多个并发Channel中使用ChannelHandler，可共享的ChannelHandler必须是线程安全的。

👆这个例子是错误的，因为它是非线程安全类。

为什么要共享一个ChannelHandler ？一个常见的原因是：构建一个单例的ChannelHandler在多个ChannelPipeline中为了跨越多个Channels来收集统计资料。

ChannelHandler、ChannelHandlerContext 补充：
如果一个ChannelHandler被标识为了’Sharable’的，并且“该ChannelHandler对象被添加到一个ChannelPipeline中多次” 或者 “该ChannelHandler对象被添加到多个ChannelPipeline中一次或者多次” 都会导致该单一ChannelHandler拥有了多个ChannelHandlerContext。
当然，如果你添加一个非共享(即，未被标示为’Sharable’)的ChannelHandle到一个ChannelPipeline多次或则添加到多个ChannelPipeline中都会导致抛出一个ChannelPipelineException异常。

异常的处理

异常处理是非常重要的部分在任何实质应用中，并且它能通过多种方式进行处理。因此，Netty提供了几种选择用于处理异常的抛出在入站或出站处理中。

处理入站异常

如果一个异常在处理一个入站事件期间被抛出，它将从被触发该异常的ChannelInboundHandler所在的位置开始流经ChannelPipeline。
Q：这句话的意思是异常也是从抛出异常的当前handler开始处理，然后传递到管道中的下一个ChannelInboundHandler？
A：是的。异常会从当前的handler开始处理，如果当前的ChannelHandler的exceptionCaught方法有调用‘ctx.fireExceptionCaught(cause);’或者‘super.exceptionCaught(ctx, cause);[其底层就是调用ctx.fireExceptionCaught(cause)]’就会将异常传递给管道中的一个ChannelInboundHandler

为了处理入站异常，你需要重写ChannelInboundHandler实现的exceptionCaught(…)方法：

因为异常将继续流进入站方向的ChannelHandler( 就和其他入站事件一样 )，实现该逻辑的ChannelInboundHandler通常位于ChannelPipeline中的最后面( 即，最后一个ChannelInboundHandler )。这能确保所有入站中的异常总能被处理，无论该异常在ChannelPipeline中的哪里发生。

如何应对一个异常可能和你的应用的具体情况而定。你可能想要关闭这个channel或者你可能试图恢复这个channel。如果你没有实现任何入站异常的处理( 或者说，没有消费任何异常 )，Netty将日志记录未处理异常的真实情况。

总结：

ChannelHandler.exceptionCaught()方法的默认实现是传递当前异常到管道中的下一个处理器中。
如果一个异常到达了管道的结尾，该异常将被记录为未处理。
你可以通过重写exceptionCaught()方法来自定义异常处理。然后你能够觉得是否要让该异常跨过该点( 即，是否需要将该异常传递到管道中的下一个处理器中 )。

处理出站异常

在出站操作中处理正常完成和异常的选项是基于下面的通知机制：

每一个出站操作将返回一个ChannelFuture。注册ChannelFutureListeners到一个ChannelFuture中，无论该事件成功与否ChannelFutureListeners在事件完成时都将得到一个通知。
几乎所有的ChannelOutboundHandler方法都会传递一个ChannelPromise实例( 作为方法的参数来传递 )。作为ChannelFuture的子类，ChannelPromise也能给指定的监听进行异步事件的通知。但是ChannelPromise还提供了可写的方法来提供即时通知：也就是说，ChannelPromise相比于ChannelFuture来说多了可以设置异步操作完成(成功或失败)的方法。

通过一个ChannelFuture实例的addListener(ChannelFutureListener)方法来添加一个ChannelFutureListener，并且有两种方式来实现这。其中一个最常见的用法是在ChannelFuture上调用addListener()方法，该ChannelFuture是一个出站操作( 如，write() )的返回值。

第二个方式是，添加一个ChannelFutureListener到一个ChannelPromise，该ChannelPromise将以参数的方式在ChannelOutboundHandler方法中传递。

6.14示例和6.13示例是等价的。但使用6.14示例的方式来处理异常不如6.13示例来的专业，而6.14示例通过自定义OutboundExceptionHandler方式处理异常虽然不专业但实现更简单。

总的来说，入站操作异常由exceptionCaught()方法处理；出站异常由ChannelFuture处理。

出站异常示例：

控制台：

👆这里是没有异常堆栈信息的。异常在ChannelFuture中持有的，我们可以通过👇的方式来获取异常。

ChannelPromise 的可写方法
通过调用ChannelPromise的setSuccess()和setFailure()方法，你能立即得知操作的状态在该ChannelHandler方法返回给调用者后。
我想这里想要表述的是，一旦调用了ChannelPromise的setSuccess()和setFailure()方法，则说明当前的异步操作就完成了。所以在ChannelPromise的setSuccess()和setFailure()方法调用后，你就能够得到异步操作的状态了。

如果你的ChannelOutboundHandler自己抛出了一个异常，那么将会发生什么？在这种情况下，Netty将会通知所有已经注册到相应ChannelPromise的监听器

后记

本文主要对Netty的ChannelHandler和ChannelPipeline进行了介绍，这两个都是是Netty中非常重要的组件，也涉及到了不少的知识点。
若文章有任何错误，望大家不吝指教:)

参考

《Netty in action》