Netty并发之ExecutionHandler

 

 

大家都知道，Netty是一个基于事件的NIO框架。在Netty中，一切网络动作都是通过事件来传播并处理的，例如：Channel读、Channel写等等。回忆下Netty的流处理模型：

Boss线程（一个服务器端口对于一个）—接收到客户端连接—生成Channel—交给Work线程池（多个Work线程）来处理。

具体的Work线程—读完已接收的数据到ChannelBuffer—触发ChannelPipeline中的ChannelHandler链来处理业务逻辑。

注意：执行ChannelHandler链的整个过程是同步的，如果业务逻辑的耗时较长，会将导致Work线程长时间被占用得不到释放，从而影响了整个服务器的并发处理能力。

所以，为了提高并发数，一般通过ExecutionHandler线程池来异步处理ChannelHandler链（worker线程在经过ExecutionHandler后就结束了，它会被ChannelFactory的worker线程池所回收）。在Netty中，只需要增加一行代码：

public ChannelPipeline getPipeline() {
         return Channels.pipeline(
                 new DatabaseGatewayProtocolEncoder(),
                 new DatabaseGatewayProtocolDecoder(),
                 executionHandler, // Must be shared
                 new DatabaseQueryingHandler());
}

例如：

ExecutionHandler executionHandler = new ExecutionHandler(
             new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576))

 

对于ExecutionHandler需要的线程池模型，Netty提供了两种可选：

1） MemoryAwareThreadPoolExecutor 通过对线程池内存的使用控制，可控制Executor中待处理任务的上限（超过上限时，后续进来的任务将被阻塞），并可控制单个Channel待处理任务的上限，防止内存溢出错误；

2） OrderedMemoryAwareThreadPoolExecutor 是 MemoryAwareThreadPoolExecutor 的子类。除了MemoryAwareThreadPoolExecutor 的功能之外，它还可以保证同一Channel中处理的事件流的顺序性，这主要是控制事件在异步处理模式下可能出现的错误的事件顺序，但它并不保证同一Channel中的事件都在一个线程中执行（通常也没必要）。

例如：

Thread X: --- Channel A (Event A1) --.   .-- Channel B (Event B2) --- Channel B (Event B3) --->
                                      \ /
                                       X
                                      / \
Thread Y: --- Channel B (Event B1) --'   '-- Channel A (Event A2) --- Channel A (Event A3) --->

上图表达的意思有几个：

（1）对整个线程池而言，处理同一个Channel的事件，必须是按照顺序来处理的。例如，必须先处理完Channel A (Event A1) ，再处理Channel A (Event A2)、Channel A (Event A3)

（2）同一个Channel的多个事件，会分布到线程池的多个线程中去处理。

（3）不同Channel的事件可以同时处理（分担到多个线程），互不影响。

OrderedMemoryAwareThreadPoolExecutor 的这种事件处理有序性是有意义的，因为通常情况下，请求发送端希望服务器能够按照顺序处理自己的请求，特别是需要多次握手的应用层协议。例如：XMPP协议。