作者：胡宗棠

来源：匠心独运的博客

在（一）篇中主要介绍了RocketMQ的协议格式，消息编解码，通信方式(同步/异步/单向)、消息发送/接收以及异步回调的主要通信流程。 而本篇将主要对RocketMQ消息队列RPC通信部分的Netty 多线程 模型进行重点介绍。

为何要使用Netty作为高性能的通信库？

在看RocketMQ的RPC通信部分时候，可能有不少同学有这样子的疑问，RocketMQ为何要选择Netty而不直接使用JDK的NIO进行网络编程呢？这里有必要先来简要介绍下Netty。 Netty是一个封装了JDK的NIO库的高性能网络通信开源框架。它提供异步的、 事件驱动 的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。

下面主要列举了下一般系统的RPC通信模块会选择Netty作为底层通信库的理由（作者认为RocketMQ的RPC同样也是基于此选择了Netty）：

（1）Netty的编程API使用简单，开发门槛低，无需编程者去关注和了解太多的NIO编程模型和概念；

（2）对于编程者来说，可根据业务的要求进行定制化地开发，通过Netty的ChannelHandler对通信框架进行灵活的定制化扩展；

（3）Netty框架本身支持拆包/解包，异常检测等机制，让编程者可以从JAVA NIO的繁琐细节中解脱，而只需要关注业务处理逻辑；

（4）Netty解决了（准确地说应该是采用了另一种方式完美规避了）JDK NIO的Bug（Epoll bug，会导致Selector空轮询，最终导致CPU 100%）；

（5）Netty框架内部对 线程 ，selector做了一些细节的优化，精心设计的reactor多线程模型，可以实现非常高效地并发处理；

（6）Netty已经在多个开源项目（Hadoop的RPC框架avro使用Netty作为通信框架）中都得到了充分验证，健壮性/可靠性比较好。

RocketMQ中RPC通信的Netty多线程模型

RocketMQ的RPC通信部分采用了”1+N+M1+M2″的Reactor多线程模式，对网络通信部分进行了一定的扩展与优化，这一节主要让我们来看下这一部分的具体设计与实现内容。

• Netty的Reactor多线程模型设计概念与简述

这里有必要先来简要介绍下Netty的Reactor多线程模型。Reactor多线程模型的设计思想是分而治之+事件驱动。

（1）分而治之

一般来说，一个网络请求连接的完整处理过程可以分为接受（accept）、数据读取（read）、解码/编码（decode/encode）、业务处理（process）、发送响应（send）这几步骤。Reactor模型将每个步骤都映射成为一个任务，服务端线程执行的最小逻辑单元不再是一次完整的网络请求，而是这个任务，且采用以非阻塞方式执行。

（2）事件驱动

每个任务对应特定网络事件。当任务准备就绪时，Reactor收到对应的网络事件通知，并将任务分发给绑定了对应网络事件的 Handler 执行。

• RocketMQ中RPC通信的1+N+M1+M2的Reactor多线程设计与实现

（1）RocketMQ中RPC通信的Reactor多线程设计与流程

RocketMQ的RPC通信采用Netty组件作为底层通信库，同样也遵循了Reactor多线程模型，同时又在这之上做了一些扩展和优化。下面先给出一张RocketMQ的RPC通信层的Netty多线程模型框架图，让大家对RocketMQ的RPC通信中的多线程分离设计有一个大致的了解。

从上面的框图中可以大致了解RocketMQ中NettyRemotingServer的Reactor 多线程模型。一个 Reactor 主线程（eventLoopGroupBoss，即为上面的1）负责监听 TCP网络连接请求，建立好连接后丢给Reactor 线程池（eventLoopGroupSelector，即为上面的“N”，源码中默认设置为3），它负责将建立好连接的 socket 注册到 selector上去（RocketMQ的源码中会自动根据OS的类型选择NIO和Epoll，也可以通过参数配置），然后监听真正的网络数据。拿到网络数据后，再丢给Worker线程池（defaultEventExecutorGroup，即为上面的“M1”，源码中默认设置为8）。

为了更为高效的处理RPC的网络请求，这里的Worker线程池是专门用于处理Netty网络通信相关的（包括编码/解码、空闲链接管理、网络连接管理以及网络请求处理）。而处理业务操作放在业务线程池中执行（这个内容在“RocketMQ的RPC通信（一）篇”中也有提到），根据 RomotingCommand 的业务请求码code去processorTable这个本地缓存变量中找到对应的 processor，然后封装成task任务后，提交给对应的业务processor处理线程池来执行（sendMessageExecutor，以发送消息为例，即为上面的 “M2”）。

下面以表格的方式列举了下上面所述的“1+N+M1+M2”Reactor多线程模型

（2）RocketMQ中RPC通信的Reactor多线程的代码具体实现

说完了Reactor多线程整体的设计与流程，大家应该就对RocketMQ的RPC通信的Netty部分有了一个比较全面的理解了，那接下来就从源码上来看下一些细节部分（在看该部分代码时候需要读者对JAVA NIO和Netty的相关概念与技术点有所了解）。在NettyRemotingServer的实例初始化时，会初始化各个相关的变量包括serverBootstrap、nettyServerConfig参数、channelEventListener监听器并同时初始化eventLoopGroupBoss和eventLoopGroupSelector两个Netty的EventLoopGroup线程池（这里需要注意的是，如果是Linux平台，并且开启了native epoll，就用EpollEventLoopGroup，这个也就是用JNI，调的c写的epoll；否则，就用Java NIO的NioEventLoopGroup。）。

具体代码如下：

1. public
3. NettyRemotingServer
4. (
5. final
7. NettyServerConfig
8. nettyServerConfig,
10. final
12. ChannelEventListener
13. channelEventListener) {
15. super
16. (nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
18. this
19. .serverBootstrap =
20. new
22. ServerBootstrap
23. ();
25. this
26. .nettyServerConfig = nettyServerConfig;
28. this
29. .channelEventListener = channelEventListener;
31. //省略部分代码
33. //初始化时候nThreads设置为1,说明RemotingServer端的Disptacher链接管理和分发请求的线程为1,用于接收客户端的TCP连接
35. this
36. .eventLoopGroupBoss =
37. new
39. NioEventLoopGroup
40. (
41. 1
42. ,
43. new
45. ThreadFactory
46. () {
48. private
50. AtomicInteger
51. threadIndex =
52. new
54. AtomicInteger
55. (
56. 0
57. );
59. @ Override
61. public
63. Thread
64. newThread(
65. Runnable
66. r) {
68. return
70. new
72. Thread
73. (r,
74. String
75. .format(
76. “NettyBoss_%d”
77. ,
78. this
79. .threadIndex.incrementAndGet()));
80. }
81. });
83. /**
84. * 根据配置设置NIO还是Epoll来作为Selector线程池
85. * 如果是Linux平台，并且开启了native epoll，就用EpollEventLoopGroup，这个也就是用JNI，调的c写的epoll；否则，就用Java NIO的NioEventLoopGroup。
86. *
87. */
89. if
90. (useEpoll()) {
92. this
93. .eventLoopGroupSelector =
94. new
96. EpollEventLoopGroup
97. (nettyServerConfig.getServerSelectorThreads(),
98. new
100. ThreadFactory
101. () {
103. private
105. AtomicInteger
106. threadIndex =
107. new
109. AtomicInteger
110. (
111. 0
112. );
114. private
116. int
117. threadTotal = nettyServerConfig.getServerSelectorThreads();
119. @Override
121. public
123. Thread
124. newThread(
125. Runnable
126. r) {
128. return
130. new
132. Thread
133. (r,
134. String
135. .format(
136. “NettyServerEPOLLSelector_%d_%d”
137. , threadTotal,
138. this
139. .threadIndex.incrementAndGet()));
140. }
141. });
142. }
143. else
144. {
146. this
147. .eventLoopGroupSelector =
148. new
150. NioEventLoopGroup
151. (nettyServerConfig.getServerSelectorThreads(),
152. new
154. ThreadFactory
155. () {
157. private
159. AtomicInteger
160. threadIndex =
161. new
163. AtomicInteger
164. (
165. 0
166. );
168. private
170. int
171. threadTotal = nettyServerConfig.getServerSelectorThreads();
173. @Override
175. public
177. Thread
178. newThread(
179. Runnable
180. r) {
182. return
184. new
186. Thread
187. (r,
188. String
189. .format(
190. “NettyServerNIOSelector_%d_%d”
191. , threadTotal,
192. this
193. .threadIndex.incrementAndGet()));
194. }
195. });
196. }
198. //省略部分代码

在NettyRemotingServer实例初始化完成后，就会将其启动。Server端在启动阶段会将之前实例化好的1个acceptor线程（eventLoopGroupBoss），N个IO线程（eventLoopGroupSelector），M1个worker 线程（defaultEventExecutorGroup）绑定上去。前面部分也已经介绍过各个线程池的作用了。

这里需要说明的是，Worker线程拿到网络数据后，就交给Netty的ChannelPipeline（其采用责任链设计模式），从Head到 Tail 的一个个Handler执行下去，这些 Handler是在创建NettyRemotingServer实例时候指定的。NettyEncoder和NettyDecoder 负责网络传输数据和 RemotingCommand 之间的编解码。NettyServerHandler 拿到解码得到的 RemotingCommand 后，根据 RemotingCommand.type 来判断是 request 还是 response来进行相应处理，根据业务请求码封装成不同的task任务后，提交给对应的业务processor处理线程池处理。

2. @Override
4. public
6. void
7. start() {
9. //默认的处理线程池组,使用默认的处理线程池组用于处理后面的多个Netty Handler的逻辑操作
11. this
12. .defaultEventExecutorGroup =
13. new
15. DefaultEventExecutorGroup
16. (
17. nettyServerConfig.getServerWorkerThreads(),
19. new
21. ThreadFactory
22. () {
24. private
26. AtomicInteger
27. threadIndex =
28. new
30. AtomicInteger
31. (
32. 0
33. );
35. @Override
37. public
39. Thread
40. newThread(
41. Runnable
42. r) {
44. return
46. new
48. Thread
49. (r,
50. “NettyServerCodecThread_”
51. +
52. this
53. .threadIndex.incrementAndGet());
54. }
55. });
57. /**
58. * 首先来看下 RocketMQ NettyServer 的 Reactor 线程模型，
59. * 一个 Reactor 主线程负责监听 TCP 连接请求;
60. * 建立好连接后丢给 Reactor 线程池，它负责将建立好连接的 socket 注册到 selector
61. * 上去（这里有两种方式，NIO和Epoll，可配置），然后监听真正的网络数据;
62. * 拿到网络数据后，再丢给 Worker 线程池;
63. *
64. */
66. //RocketMQ-> Java NIO的1+N+M模型：1个acceptor线程，N个IO线程，M1个worker 线程。
68. ServerBootstrap
69. childHandler =
71. this
72. .serverBootstrap.
73. group
74. (
75. this
76. .eventLoopGroupBoss,
77. this
78. .eventLoopGroupSelector)
79. .channel(useEpoll() ?
80. EpollServerSocketChannel
81. .
82. class
83. :
84. NioServerSocketChannel
85. .
86. class
87. )
88. .option(
89. ChannelOption
90. .SO_BACKLOG,
91. 1024
92. )
94. //服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小
95. .option(
96. ChannelOption
97. .SO_REUSEADDR,
98. true
99. )
100. //这个参数表示允许重复使用本地地址和端口
101. .option(
102. ChannelOption
103. .SO_KEEPALIVE,
104. false
105. )
106. //当设置该选项以后，如果在两小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文。
107. .childOption(
108. ChannelOption
109. .TCP_NODELAY,
110. true
111. )
112. //该参数的作用就是禁止使用Nagle算法，使用于小数据即时传输
113. .childOption(
114. ChannelOption
115. .SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
116. //这两个参数用于操作接收缓冲区和发送缓冲区
117. .childOption(
118. ChannelOption
119. .SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
120. .localAddress(
121. new
123. InetSocketAddress
124. (
125. this
126. .nettyServerConfig.getListenPort()))
127. .childHandler(
128. new
130. ChannelInitializer
131. <
132. SocketChannel
133. >() {
135. @Override
137. public
139. void
140. initChannel(
141. SocketChannel
142. ch)
143. throws
145. Exception
146. {
147. ch.pipeline()
148. .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
150. new
152. HandshakeHandler
153. (
154. TlsSystemConfig
155. .tlsMode))
156. .addLast(defaultEventExecutorGroup,
158. new
160. NettyEncoder
161. (),
162. //rocketmq解码器,他们分别覆盖了父类的encode和decode方法
164. new
166. NettyDecoder
167. (),
168. //rocketmq编码器
170. new
172. IdleStateHandler
173. (
174. 0
175. ,
176. 0
177. , nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
178. //Netty自带的心跳管理器
180. new
182. NettyConnectManageHandler
183. (),
184. //连接管理器，他负责捕获新连接、连接断开、异常等事件，然后统一调度到NettyEventExecuter处理器处理。
186. new
188. NettyServerHandler
189. ()
190. //当一个消息经过前面的解码等步骤后，然后调度到channelRead0方法，然后根据消息类型进行分发
191. );
192. }
193. });
195. if
196. (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
197. childHandler.childOption(
198. ChannelOption
199. .ALLOCATOR,
200. PooledByteBufAllocator
201. .DEFAULT);
202. }
204. try
205. {
207. ChannelFuture
208. sync =
209. this
210. .serverBootstrap.bind().sync();
212. InetSocketAddress
213. addr = (
214. InetSocketAddress
215. ) sync.channel().localAddress();
217. this
218. .port = addr.getPort();
219. }
220. catch
221. (
222. InterruptedException
223. e1) {
225. throw
227. new
229. RuntimeException
230. (
231. “this.serverBootstrap.bind().sync() InterruptedException”
232. , e1);
233. }
235. if
236. (
237. this
238. .channelEventListener !=
239. null
240. ) {
242. this
243. .nettyEventExecutor.start();
244. }
246. //定时扫描responseTable,获取返回结果,并且处理超时
248. this
249. .timer.scheduleAtFixedRate(
250. new
252. TimerTask
253. () {
255. @Override
257. public
259. void
260. run() {
262. try
263. {
265. NettyRemotingServer
266. .
267. this
268. .scanResponseTable();
269. }
270. catch
271. (
272. Throwable
273. e) {
274. log.error(
275. “scanResponseTable exception”
276. , e);
277. }
278. }
279. },
280. 1000
281. *
282. 3
283. ,
284. 1000
285. );
286. }

从上面的描述中可以概括得出RocketMQ的RPC通信部分的Reactor线程池模型框图。

整体可以看出RocketMQ的RPC通信借助Netty的多线程模型，其服务端监听线程和IO线程分离，同时将RPC通信层的业务逻辑与处理具体业务的线程进一步相分离。时间可控的简单业务都直接放在RPC通信部分来完成，复杂和时间不可控的业务提交至后端业务线程池中处理，这样提高了通信效率和MQ整体的性能。（ps：其中抽象出NioEventLoop来表示一个不断循环执行处理任务的线程，每个NioEventLoop有一个selector，用于监听绑定在其上的socket链路。）

总结

仔细阅读RocketMQ的过程中收获了很多关于网络通信设计技术和知识点。对于刚接触开源版的RocketMQ的童鞋来说，想要自己掌握RPC通信部分的各个技术知识点，还需要不断地使用本地环境进行debug调试和阅读源码反复思考。限于笔者的才疏学浅，对本文内容可能还有理解不到位的地方，如有阐述不合理之处还望留言一起探讨。