作者：胡宗棠

来源： 匠心独运的博客

文章摘要：借用小厮的一句话“消息队列的本质在于消息的发送、存储和接收”。那么，对于一款消息队列来说，如何做到消息的高效发送与接收是重点和关键。

一、RocketMQ中Remoting通信模块概览

RocketMQ消息队列的整体部署架构如下图所示：

先来说下RocketMQ消息队列集群中的几个角色：

（1）NameServer：在MQ集群中做的是做命名服务，更新和路由发现 broker服务；

（2）Broker-Master：broker 消息主机服务器；

（3）Broker-Slave：broker 消息从机服务器；

（4）Producer：消息生产者；

（5）Consumer：消息消费者；

其中，RocketMQ集群的一部分通信如下：

（1）Broker启动后需要完成一次将自己注册至NameServer的操作；随后每隔30s时间定期向NameServer上报Topic路由信息；

（2）消息生产者Producer作为客户端发送消息时候，需要根据Msg的Topic从本地缓存的TopicPublishInfoTable获取路由信息。如果没有则更新路由信息会从NameServer上重新拉取；

（3）消息生产者Producer根据（2）中获取的路由信息选择一个队列（MessageQueue）进行消息发送；Broker作为消息的接收者收消息并落盘存储； 从上面（1）~（3）中可以看出在消息生产者, Broker和Name Server 之间都会发生通信（这里只说了MQ的部分通信），因此如何设计一个良好的网络通信模块在MQ中至关重要，它将决定RocketMQ集群整体的消息传输能力与最终的性能。

rocketmq-remoting 模块是 RocketMQ消息队列中负责网络通信的模块，它几乎被其他所有需要网络通信的模块（诸如rocketmq-client、rocketmq-server、rocketmq-namesrv）所依赖和引用。为了实现客户端与服务器之间高效的数据请求与接收，RocketMQ消息队列自定义了通信协议并在 Netty 的基础之上扩展了通信模块。ps:鉴于RocketMQ的通信模块是建立在Netty基础之上的，因此在阅读RocketMQ的源码之前，读者最好先对Netty的多线程模型、JAVA NIO模型均有一定的了解，这样子理解RocketMQ源码会较为快一些。

作者阅读的RocketMQ版本是4.2.0, 依赖的netty版本是4.0.42.Final. RocketMQ的代码结构图如下:

源码部分主要可以分为rocketmq-broker，rocketmq-client，rocketmq-common，rocketmq-filterSrv，rocketmq-namesrv和rocketmq-remoting等模块，通信框架就封装在rocketmq-remoting模块中。 本文主要从RocketMQ的协议格式，消息编解码，通信方式(同步/异步/单向)和具体的发送/接收消息的通信流程来进行阐述等。

二、RocketMQ中Remoting通信模块的具体实现

• Remoting通信模块的类结构图

从类层次结构来看：

（1）RemotingService：为最上层的接口，提供了三个方法：

1. void
2. start();
3. void
4. shutdown();
5. void
6. register RPC Hook(
7. RPCHook
8. rpcHook);

（2）RemotingClient/RemotingSever：两个接口继承了最上层接口—RemotingService，分别各自为Client和Server提供所必需的方法，下面所列的是RemotingServer的方法：

1. /**
2. * 同RemotingClient端一样
3. *
4. * @param requestCode
5. * @param processor
6. * @param executor
7. */
9. void
10. registerProcessor(
11. final
13. int
14. requestCode,
15. final
17. NettyRequestProcessor
18. processor,
20. final
22. ExecutorService
23. executor);
25. /**
26. * 注册默认的处理器
27. *
28. * @param processor
29. * @param executor
30. */
32. void
33. registerDefaultProcessor(
34. final
36. Netty request Processor
37. processor,
38. final
40. ExecutorService
41. executor);
43. int
44. localListenPort();
46. /**
47. * 根据请求code来获取不同的处理Pair
48. *
49. * @param requestCode
50. * @return
51. */
53. Pair
54. <
55. NettyRequestProcessor
56. ,
57. ExecutorService
58. > getProcessorPair(
59. final
61. int
62. requestCode);
64. /**
65. * 同RemotingClient端一样,同步通信,有返回RemotingCommand
66. * @param channel
67. * @param request
68. * @param timeoutMillis
69. * @return
70. * @ throws InterruptedException
71. * @throws RemotingSendRequestException
72. * @throws RemotingTimeoutException
73. */
75. RemotingCommand
76. invokeSync(
77. final
79. Channel
80. channel,
81. final
83. RemotingCommand
84. request,
86. final
88. long
89. timeoutMillis)
90. throws
92. InterruptedException
93. ,
94. RemotingSendRequestException
95. ,
97. RemotingTimeoutException
98. ;
100. /**
101. * 同RemotingClient端一样,异步通信,无返回RemotingCommand
102. *
103. * @param channel
104. * @param request
105. * @param timeoutMillis
106. * @param invokeCallback
107. * @throws InterruptedException
108. * @throws RemotingTooMuchRequestException
109. * @throws RemotingTimeoutException
110. * @throws RemotingSendRequestException
111. */
113. void
114. invokeAsync(
115. final
117. Channel
118. channel,
119. final
121. RemotingCommand
122. request,
123. final
125. long
126. timeoutMillis,
128. final
130. InvokeCallback
131. invokeCallback)
132. throws
134. InterruptedException
135. ,
137. RemotingTooMuchRequestException
138. ,
139. RemotingTimeoutException
140. ,
141. RemotingSendRequestException
142. ;
144. /**
145. * 同RemotingClient端一样，单向通信，诸如心跳包
146. *
147. * @param channel
148. * @param request
149. * @param timeoutMillis
150. * @throws InterruptedException
151. * @throws RemotingTooMuchRequestException
152. * @throws RemotingTimeoutException
153. * @throws RemotingSendRequestException
154. */
156. void
157. invokeOneway(
158. final
160. Channel
161. channel,
162. final
164. RemotingCommand
165. request,
166. final
168. long
169. timeoutMillis)
171. throws
173. InterruptedException
174. ,
175. RemotingTooMuchRequestException
176. ,
177. RemotingTimeoutException
178. ,
180. RemotingSendRequestException
181. ;

（3）NettyRemotingAbstract：Netty通信处理的 抽象类 ，定义并封装了Netty处理的公共处理方法；

（4）NettyRemotingClient以及NettyRemotingServer：分别实现了RemotingClient和RemotingServer, 都继承了NettyRemotingAbstract抽象类。RocketMQ中其他的组件（如client、nameServer、broker在进行消息的发送和接收时均使用这两个组件）

• 消息的协议设计与编码解码

在Client和Server之间完成一次消息发送时，需要对发送的消息进行一个协议约定，因此就有必要自定义RocketMQ的消息协议。同时，为了高效地在网络中传输消息和对收到的消息读取，就需要对消息进行编解码。在RocketMQ中，RemotingCommand这个类在消息传输过程中对所有数据内容的封装，不但包含了所有的数据结构，还包含了编码解码操作。

RemotingCommand类的部分成员变量如下：

Header字段类型Request说明Response说明codeint请求操作码，应答方根据不同的请求码进行不同的业务处理应答响应码。0表示成功，非0则表示各种错误languageLanguageCode请求方实现的语言应答方实现的语言versionint请求方程序的版本应答方程序的版本opaqueint相当于reqeustId，在同一个连接上的不同请求标识码，与响应消息中的相对应应答不做修改直接返回flagint区分是普通RPC还是onewayRPC得标志区分是普通RPC还是onewayRPC得标志remarkString传输自定义文本信息传输自定义文本信息extFieldsHashMap请求自定义扩展信息响应自定义扩展信息

这里展示下Broker向NameServer发送一次心跳注册的报文：

1. [
2. code=
3. 103
4. ,
5. //这里的103对应的code就是broker向nameserver注册自己的消息
6. language=JAVA,
7. version=
8. 137
9. ,
10. opaque=
11. 58
12. ,
13. //这个就是requestId
14. flag(B)=
15. 0
16. ,
17. remark=
18. null
19. ,
20. extFields={
21. brokerId=
22. 0
23. ,
24. clusterName=
25. DefaultCluster
26. ,
27. brokerAddr=ip1:
28. 10911
29. ,
30. haServerAddr=ip1:
31. 10912
32. ,
33. brokerName=LAPTOP-SMF2CKDN
34. },
35. serializeTypeCurrentRPC=JSON

下面来看下RocketMQ通信协议的格式：

可以看到传输内容主要可以分为以下4部分：

（1）消息长度： 总长度，四个字节存储，占用一个int类型；

（2）序列化类型&消息头长度： 同样占用一个int类型，第一个字节表示序列化类型，后面三个字节表示消息头长度；

（3）消息头数据： 经过序列化后的消息头数据；

（4）消息主体数据： 消息主体的二进制字节数据内容；

消息的编码和解码分别在RemotingCommand类的encode和decode方法中完成，下面是消息编码encode方法的具体实现：

1. public
3. byte Buffer
4. encode() {
6. // 1> header length size
8. int
9. length =
10. 4
11. ;
12. //消息总长度
14. // 2> header data length
16. //将消息头编码成byte[]
18. byte
19. [] headerData =
20. this
21. .headerEncode();
23. //计算头部长度
24. length += headerData.length;
26. // 3> body data length
28. if
29. (
30. this
31. .body !=
32. null
33. ) {
35. //消息主体长度
36. length += body.length;
37. }
39. //分配ByteBuffer, 这边加了4,
41. //这是因为在消息总长度的计算中没有将存储头部长度的4个字节计算在内
43. ByteBuffer
44. result =
45. ByteBuffer
46. .allocate(
47. 4
48. + length);
50. // length
52. //将消息总长度放入ByteBuffer
53. result.putInt(length);
55. // header length
57. //将消息头长度放入ByteBuffer
58. result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
60. // header data
62. //将消息头数据放入ByteBuffer
63. result.put(headerData);
65. // body data;
67. if
68. (
69. this
70. .body !=
71. null
72. ) {
74. //将消息主体放入ByteBuffer
75. result.put(
76. this
77. .body);
78. }
80. //重置ByteBuffer的position位置
81. result.flip();
83. return
84. result;
85. }
87. /**
88. * markProtocolType方法是将RPC类型和headerData长度编码放到一个byte[4]数组中
89. *
90. * @param source
91. * @param type
92. * @return
93. */
95. public
97. static
99. byte
100. [] markProtocolType(
101. int
102. source,
103. SerializeType
104. type) {
106. byte
107. [] result =
108. new
110. byte
111. [
112. 4
113. ];
114. result[
115. 0
116. ] = type.getCode();
118. //右移16位后再和255与->“16-24位”
119. result[
120. 1
121. ] = (
122. byte
123. ) ((source >>
124. 16
125. ) &
126. 0xFF
127. );
129. //右移8位后再和255与->“8-16位”
130. result[
131. 2
132. ] = (
133. byte
134. ) ((source >>
135. 8
136. ) &
137. 0xFF
138. );
140. //右移0位后再和255与->“8-0位”
141. result[
142. 3
143. ] = (
144. byte
145. ) (source &
146. 0xFF
147. );
149. return
150. result;
151. }

消息解码decode方法是编码的逆向过程，其具体实现如下：

1. public
3. static
5. RemotingCommand
6. decode(
7. final
9. ByteBuffer
10. byteBuffer) {
12. //获取byteBuffer的总长度
14. int
15. length = byteBuffer.limit();
17. //获取前4个字节，组装int类型，该长度为总长度
19. int
20. oriHeaderLen = byteBuffer.getInt();
22. //获取消息头的长度，这里和0xFFFFFF做与运算，编码时候的长度即为24位
24. int
25. headerLength = getHeaderLength(oriHeaderLen);
27. byte
28. [] headerData =
29. new
31. byte
32. [headerLength];
33. byteBuffer.
34. get
35. (headerData);
37. RemotingCommand
38. cmd = headerDecode(headerData, getProtocolType(oriHeaderLen));
40. int
41. bodyLength = length –
42. 4
43. – headerLength;
45. byte
46. [] bodyData =
47. null
48. ;
50. if
51. (bodyLength >
52. 0
53. ) {
54. bodyData =
55. new
57. byte
58. [bodyLength];
59. byteBuffer.
60. get
61. (bodyData);
62. }
63. cmd.body = bodyData;
65. return
66. cmd;
67. }

• 消息的通信方式和通信流程

在RocketMQ消息队列中支持通信的方式主要有以下三种：

（1）同步(sync)

（2）异步(async)

（3）单向(oneway)

其中“同步”通信模式相对简单，一般用在发送心跳包场景下，无需关注其Response。本文将主要介绍RocketMQ的异步通信流程（限于篇幅，读者可以按照同样的模式进行分析同步通信流程）。下面先给出了RocketMQ异步通信的整体流程图：

下面两小节内容主要介绍了Client端发送请求消息、Server端接收消息的具体实现并简要分析的Client端的回调。

1）Client发送请求消息的具体实现

当客户端调用异步通信接口—invokeAsync时候，先由RemotingClient的实现类—NettyRemotingClient根据addr获取相应的channel（如果本地缓存中没有则创建），随后调用invokeAsyncImpl方法，将数据流转给抽象类NettyRemotingAbstract处理（真正做完发送请求动作的是在NettyRemotingAbstract抽象类的invokeAsyncImpl方法里面）。

具体发送请求消息的源代码如下所示：

2. /**
3. * invokeAsync（异步调用）
4. *
5. */
7. public
9. void
10. invokeAsyncImpl(
11. final
13. Channel
14. channel,
15. final
17. RemotingCommand
18. request,
19. final
21. long
22. timeoutMillis,
24. final
26. InvokeCallback
27. invokeCallback)
29. throws
31. InterruptedException
32. ,
33. RemotingTooMuchRequestException
34. ,
35. RemotingTimeoutException
36. ,
37. RemotingSendRequestException
38. {
40. //相当于request ID, RemotingCommand会为每一个request产生一个request ID, 从0开始, 每次加1
42. final
44. int
45. opaque = request.getOpaque();
47. boolean
48. acquired =
49. this
50. .semaphoreAsync.tryAcquire(timeoutMillis,
51. TimeUnit
52. .MILLISECONDS);
54. if
55. (acquired) {
57. final
59. SemaphoreReleaseOnlyOnce
60. once =
61. new
63. SemaphoreReleaseOnlyOnce
64. (
65. this
66. .semaphoreAsync);
68. //根据request ID构建ResponseFuture
70. final
72. ResponseFuture
73. responseFuture =
74. new
76. ResponseFuture
77. (opaque, timeoutMillis, invokeCallback, once);
79. //将ResponseFuture放入responseTable
81. this
82. .responseTable.put(opaque, responseFuture);
84. try
85. {
87. //使用Netty的channel发送请求数据
88. channel.writeAndFlush(request).addListener(
89. new
91. ChannelFutureListener
92. () {
94. //消息发送后执行
96. @Override
98. public
100. void
101. operationComplete(
102. ChannelFuture
103. f)
104. throws
106. Exception
107. {
109. if
110. (f.isSuccess()) {
112. //如果发送消息成功给Server，那么这里直接Set后return
113. responseFuture.setSendRequestOK(
114. true
115. );
117. return
118. ;
119. }
120. else
121. {
122. responseFuture.setSendRequestOK(
123. false
124. );
125. }
126. responseFuture.putResponse(
127. null
128. );
129. responseTable.remove(opaque);
131. try
132. {
134. //执行回调
135. executeInvokeCallback(responseFuture);
136. }
137. catch
138. (
139. Throwable
140. e) {
141. log .warn(
142. “excute callback in writeAndFlush addListener, and callback throw”
143. , e);
144. }
145. finally
146. {
148. //释放信号量
149. responseFuture.release();
150. }
151. log.warn(
152. “send a request command to channel <{}> failed.”
153. ,
154. RemotingHelper
155. .parseChannelRemoteAddr(channel));
156. }
157. });
158. }
159. catch
160. (
161. Exception
162. e) {
164. //异常处理
165. responseFuture.release();
166. log.warn(
167. “send a request command to channel <”
168. +
169. RemotingHelper
170. .parseChannelRemoteAddr(channel) +
171. “> Exception”
172. , e);
174. throw
176. new
178. RemotingSendRequestException
179. (
180. RemotingHelper
181. .parseChannelRemoteAddr(channel), e);
182. }
183. }
184. else
185. {
187. if
188. (timeoutMillis <=
189. 0
190. ) {
192. throw
194. new
196. RemotingTooMuchRequestException
197. (
198. “invokeAsyncImpl invoke too fast”
199. );
200. }
201. else
202. {
204. String
205. info =
207. String
208. .format(
209. “invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d”
210. ,
211. timeoutMillis,
213. this
214. .semaphoreAsync.getQueueLength(),
216. this
217. .semaphoreAsync.availablePermits()
218. );
219. log.warn(info);
221. throw
223. new
225. RemotingTimeoutException
226. (info);
227. }
228. }
229. }

在Client端发送请求消息时有个比较重要的数据结构需要注意下：

（1）responseTable—保存请求码与响应关联映射

1. protected
3. final
5. ConcurrentHashMap
6. <
7. Integer
9. /* opaque */
10. ,
11. ResponseFuture
12. > responseTable

opaque表示请求发起方在同个连接上不同的请求标识代码，每次发送一个消息的时候，可以选择同步阻塞/异步非阻塞的方式。无论是哪种通信方式，都会保存请求操作码至ResponseFuture的Map映射—responseTable中。

（2）ResponseFuture—保存返回响应（包括回调执行方法和信号量）

1. public
3. ResponseFuture
4. (
5. int
6. opaque,
7. long
8. timeoutMillis,
9. InvokeCallback
10. invokeCallback,
12. SemaphoreReleaseOnlyOnce
13. once) {
15. this
16. .opaque = opaque;
18. this
19. .timeoutMillis = timeoutMillis;
21. this
22. .invokeCallback = invokeCallback;
24. this
25. .once = once;
26. }

对于同步通信来说，第三、四个参数为null；而对于异步通信来说，invokeCallback是在收到消息响应的时候能够根据responseTable找到请求码对应的回调执行方法，semaphore参数用作流控，当多个线程同时往一个连接写数据时可以通过信号量控制permit同时写许可的数量。

（3）异常发送流程处理—定时扫描responseTable本地缓存

在发送消息时候，如果遇到异常情况（比如服务端没有response返回给客户端或者response因网络而丢失），上面所述的responseTable的本地缓存Map将会出现堆积情况。这个时候需要一个定时任务来专门做responseTable的清理回收。在RocketMQ的客户端/服务端启动时候会产生一个频率为1s调用一次来的定时任务检查所有的responseTable缓存中的responseFuture变量，判断是否已经得到返回, 并进行相应的处理。

1. public
3. void
4. scanResponseTable() {
6. final
8. List
9. <
10. ResponseFuture
11. > rfList =
12. new
14. LinkedList
15. <
16. ResponseFuture
17. >();
19. Iterator
20. <
21. Entry
22. <
23. Integer
24. ,
25. ResponseFuture
26. >> it =
27. this
28. .responseTable.entrySet().iterator();
30. while
31. (it.hasNext()) {
33. Entry
34. <
35. Integer
36. ,
37. ResponseFuture
38. >
39. next
40. = it.
41. next
42. ();
44. ResponseFuture
45. rep =
46. next
47. .getValue();
49. if
50. ((rep.getBeginTimestamp() + rep.getTimeoutMillis() +
51. 1000
52. ) <=
53. System
54. .currentTimeMillis()) {
55. rep.release();
56. it.remove();
57. rfList.add(rep);
58. log.warn(
59. “remove timeout request, ”
60. + rep);
61. }
62. }
64. for
65. (
66. ResponseFuture
67. rf : rfList) {
69. try
70. {
71. executeInvokeCallback(rf);
72. }
73. catch
74. (
75. Throwable
76. e) {
77. log.warn(
78. “scanResponseTable, operationComplete Exception”
79. , e);
80. }
81. }
82. }

2）Server端接收消息并进行处理的具体实现

Server端接收消息的处理入口在NettyServerHandler类的channelRead0方法中，其中调用了processMessageReceived方法（这里省略了Netty服务端消息流转的大部分流程和逻辑）。

其中服务端最为重要的处理请求方法实现如下：

1. public
3. void
4. processRequestCommand(
5. final
7. ChannelHandlerContext
8. ctx,
9. final
11. RemotingCommand
12. cmd) {
14. //根据RemotingCommand中的code获取processor和ExecutorService
16. final
18. Pair
19. <
20. NettyRequestProcessor
21. ,
22. ExecutorService
23. > matched =
24. this
25. .processorTable.
26. get
27. (cmd.getCode());
29. final
31. Pair
32. <
33. NettyRequestProcessor
34. ,
35. ExecutorService
36. > pair =
37. null
38. == matched ?
39. this
40. .defaultRequestProcessor : matched;
42. final
44. int
45. opaque = cmd.getOpaque();
47. if
48. (pair !=
49. null
50. ) {
52. Runnable
53. run =
54. new
56. Runnable
57. () {
59. @Override
61. public
63. void
64. run() {
66. try
67. {
69. //rpc hook
71. RPCHook
72. rpcHook =
73. NettyRemotingAbstract
74. .
75. this
76. .getRPCHook();
78. if
79. (rpcHook !=
80. null
81. ) {
82. rpcHook.doBeforeRequest(
83. RemotingHelper
84. .parseChannelRemoteAddr(ctx.channel()), cmd);
85. }
87. //processor处理请求
89. final
91. RemotingCommand
92. response = pair.getObject1().processRequest(ctx, cmd);
94. //rpc hook
96. if
97. (rpcHook !=
98. null
99. ) {
100. rpcHook.doAfterResponse(
101. RemotingHelper
102. .parseChannelRemoteAddr(ctx.channel()), cmd, response);
103. }
105. if
106. (!cmd.isOnewayRPC()) {
108. if
109. (response !=
110. null
111. ) {
112. response.setOpaque(opaque);
113. response.markResponseType();
115. try
116. {
117. ctx.writeAndFlush(response);
118. }
119. catch
120. (
121. Throwable
122. e) {
123. PLOG.error(
124. “process request over, but response failed”
125. , e);
126. PLOG.error(cmd. toString ());
127. PLOG.error(response.toString());
128. }
129. }
130. else
131. {
132. }
133. }
134. }
135. catch
136. (
137. Throwable
138. e) {
140. if
141. (!
142. “com.aliyun.openservices.ons.api.impl.authority.exception.AuthenticationException”
143. .equals(e.getClass().getCanonicalName())) {
144. PLOG.error(
145. “process request exception”
146. , e);
147. PLOG.error(cmd.toString());
148. }
150. if
151. (!cmd.isOnewayRPC()) {
153. final
155. RemotingCommand
156. response =
157. RemotingCommand
158. .createResponseCommand(
159. RemotingSysResponseCode
160. .SYSTEM_ERROR,
161. //
163. RemotingHelper
164. .exceptionSimpleDesc(e));
165. response.setOpaque(opaque);
166. ctx.writeAndFlush(response);
167. }
168. }
169. }
170. };
172. if
173. (pair.getObject1().rejectRequest()) {
175. final
177. RemotingCommand
178. response =
179. RemotingCommand
180. .createResponseCommand(
181. RemotingSysResponseCode
182. .SYSTEM_BUSY,
184. “[REJECTREQUEST]system busy, start flow control for a while”
185. );
186. response.setOpaque(opaque);
187. ctx.writeAndFlush(response);
189. return
190. ;
191. }
193. try
194. {
196. //封装requestTask
198. final
200. RequestTask
201. requestTask =
202. new
204. RequestTask
205. (run, ctx.channel(), cmd);
207. //想线程池提交requestTask
208. pair.getObject2().submit(requestTask);
209. }
210. catch
211. (
212. RejectedExecutionException
213. e) {
215. if
216. ((
217. System
218. .currentTimeMillis() %
219. 10000
220. ) ==
221. 0
222. ) {
223. PLOG.warn(
224. RemotingHelper
225. .parseChannelRemoteAddr(ctx.channel())
226. //
227. +
228. “, too many requests and system thread pool busy, RejectedExecutionException ”
230. //
231. + pair.getObject2().toString()
232. //
233. +
234. ” request code: ”
235. + cmd.getCode());
236. }
238. if
239. (!cmd.isOnewayRPC()) {
241. final
243. RemotingCommand
244. response =
245. RemotingCommand
246. .createResponseCommand(
247. RemotingSysResponseCode
248. .SYSTEM_BUSY,
250. “[OVERLOAD]system busy, start flow control for a while”
251. );
252. response.setOpaque(opaque);
253. ctx.writeAndFlush(response);
254. }
255. }
256. }
257. else
258. {
260. String
261. error =
262. ” request type ”
263. + cmd.getCode() +
264. ” not supported”
265. ;
267. //构建response
269. final
271. RemotingCommand
272. response =
274. RemotingCommand
275. .createResponseCommand(
276. RemotingSysResponseCode
277. .REQUEST_CODE_NOT_SUPPORTED, error);
278. response.setOpaque(opaque);
279. ctx.writeAndFlush(response);
280. PLOG.error(
281. RemotingHelper
282. .parseChannelRemoteAddr(ctx.channel()) + error);
283. }
284. }

上面的请求处理方法中根据RemotingCommand的请求业务码来匹配到相应的业务处理器；然后生成一个新的线程提交至对应的业务线程池进行异步处理。

（1）processorTable—请求业务码与业务处理、业务线程池的映射变量

2. protected
4. final
6. HashMap
7. <
8. Integer
9. /* request code */
10. ,
11. Pair
12. <
13. NettyRequestProcessor
14. ,
15. ExecutorService
16. >> processorTable =
18. new
20. HashMap
21. <
22. Integer
23. ,
24. Pair
25. <
26. NettyRequestProcessor
27. ,
28. ExecutorService
29. >>(
30. 64
31. );

个人觉得RocketMQ这种做法是为了给不同类型的请求业务码指定不同的处理器Processor处理，同时消息实际的处理并不是在当前线程，而是被封装成task放到业务处理器Processor对应的线程池中完成异步执行。(在RocketMQ中能看到很多地方都是这样的处理，这样的设计能够最大程度的保证异步，保证每个线程都专注处理自己负责的东西）

3）Client端异步回调执行的实现分析

看到这里可能有一些同学会疑问Client端的异步回调究竟在哪里执行的？从上面“RocketMQ异步通信的整体时序图”来看，回调执行处理的流程的确是放在了Client端来完成，而rocketmq-remoting通信模块中只是给异步回调处理提供了接口。

这里可以看下rocketmq-client模块异步发送消息的部分代码（限于篇幅也只是列举了异步回调执行的部分代码）：

1. private
3. void
4. sendMessageAsync(
6. final
8. String
9. addr,
11. final
13. String
14. brokerName,
16. final
18. Message
19. msg,
21. final
23. long
24. timeoutMillis,
26. final
28. RemotingCommand
29. request,
31. final
33. SendCallback
34. sendCallback,
36. final
38. TopicPublishInfo
39. topicPublishInfo,
41. final
43. MQClientInstance
44. instance,
46. final
48. int
49. retryTimesWhenSendFailed,
51. final
53. AtomicInteger
54. times,
56. final
58. SendMessageContext
59. context,
61. final
63. DefaultMQProducerImpl
64. producer
65. )
66. throws
68. InterruptedException
69. ,
70. RemotingException
71. {
73. this
74. .remotingClient.invokeAsync(addr, request, timeoutMillis,
75. new
77. InvokeCallback
78. () {
80. @Override
82. public
84. void
85. operationComplete(
86. ResponseFuture
87. responseFuture) {
89. //先从Server端返回的responseFuture变量中获取RemotingCommand的值
91. RemotingCommand
92. response = responseFuture.getResponseCommand();
94. if
95. (
96. null
97. == sendCallback && response !=
98. null
99. ) {
101. try
102. {
104. //Client端处理发送消息的Reponse返回（包括对消息返回体的头部进行解码，
106. //取得“topic”、“BrokerName”、“QueueId”等值）
108. //随后构建sendResult对象并设置Context上下文中
110. SendResult
111. sendResult =
112. MQClientAPIImpl
113. .
114. this
115. .processSendResponse(brokerName, msg, response);
117. if
118. (context !=
119. null
120. && sendResult !=
121. null
122. ) {
123. context.setSendResult(sendResult);
124. context.getProducer().executeSendMessageHookAfter(context);
125. }
126. }
127. catch
128. (
129. Throwable
130. e) {
131. }
132. producer.updateFaultItem(brokerName,
133. System
134. .currentTimeMillis() – responseFuture.getBeginTimestamp(),
135. false
136. );
138. return
139. ;
140. }
142. //省略其他部分代码
144. //……
145. }

这里需要结合3.1节的内容和NettyRemotingAbstract抽象类的processResponseCommand方法，便可以明白Client端实现异步回调的大致流程了。在Client端发送异步消息时候（rocketmq-client模块最终调用sendMessageAsync方法时），会将InvokeCallback的接口注入，而在Server端的异步线程由上面所讲的业务线程池真正执行后，返回response给Client端时候才会去触发执行。

NettyRemotingAbstract抽象类的processResponseCommand方法的具体代码如下：

1. public
3. void
4. processResponseCommand(
5. ChannelHandlerContext
6. ctx,
7. RemotingCommand
8. cmd) {
10. //从RemotingCommand中获取opaque值
12. final
14. int
15. opaque = cmd.getOpaque();‘
17. //从本地缓存的responseTable这个Map中取出本次异步通信连接对应的ResponseFuture变量
19. final
21. ResponseFuture
22. responseFuture = responseTable.
23. get
24. (opaque);
26. if
27. (responseFuture !=
28. null
29. ) {
30. responseFuture.setResponseCommand(cmd);
31. responseTable.remove(opaque);
33. if
34. (responseFuture.getInvokeCallback() !=
35. null
36. ) {
38. //在这里真正去执行Client注入进来的异步回调方法
39. executeInvokeCallback(responseFuture);
40. }
41. else
42. {
44. //否则释放responseFuture变量
45. responseFuture.putResponse(cmd);
46. responseFuture.release();
47. }
48. }
49. else
50. {
51. log.warn(
52. “receive response, but not matched any request, ”
53. +
54. RemotingHelper
55. .parseChannelRemoteAddr(ctx.channel()));
56. log.warn(cmd.toString());
57. }
58. }

三、总结

刚开始看RocketMQ源码—RPC通信模块可能觉得略微有点复杂，但是只要能够抓住Client端发送请求消息、Server端接收消息并处理的流程以及回调过程来分析和梳理，那么整体来说并不复杂。RPC通信部分也是RocketMQ源码中最重要的部分之一，想要对其中的全过程和细节有更为深刻的理解，还需要多在本地环境Debug和分析对应的日志。同时，鉴于篇幅所限，本篇还没有来得及对RocketMQ的Netty多线程模型进行介绍，将在消息中间件—RocketMQ的RPC通信（二）篇中来做详细地介绍。