分布式系统各个角色间的通信效率很关键，通信效率的高低直接影响系统性能，基于Socket实现一个高效的TCP通信协议是很有挑战的，本节介绍RocketMQ是如何解决这个问题的。
1.Remoting模块
RocketMQ的通信相关代码在Remoting模块里，先来看看主要类结构，如图4-1所示。
图4-1 Remoting模块的类继承关系
RemotingService为最上层接口，定义了三个方法：
·void start（）；
·void shutdown（）；
·void registerRPCHook（RPCHook rpcHook）；
RemotingClient和RemotingServer继承RemotingService接口，并增加了自己特有的方法。RemotingClient的主要函数定义如代码清单4-6所示。
代码清单4-6 RemotingClient主要函数定义
void registerProcessor(final int requestCode, final NettyRequestProcessor processor,final ExecutorService executor);
RemotingCommand invokeSync(final String addr, final RemotingCommand request, final long timeoutMillis);
void invokeAsync(final String addr, final RemotingCommand request, final long timeoutMillis,final InvokeCallback invokeCallback);
void invokeOneway(final String addr, final RemotingCommand request, final long timeoutMillis);
void updateNameServerAddressList(final List<String> addrs);
然后看看具体的实现类，NettyRemotingClient和NettyRemotingServer分别实现了RemotingClient和RemotingServer，而且都继承了NettyRemotingAbstract类。通过上面的封装，RocketMQ各个模块间的通信，可以通过发送统一格式的自定义消息（RemotingCommand）来完成，各个模块间的通信实现简洁明了。比如NameServer模块中，NameServerController有一个remotingServer变量，NameServer在启动时初始化各个变量，然后启动remotingServer即可，剩下NameServer要做的是专心实现处理RemotingCommand的逻辑，如代码清单4-7所示。
代码清单4-7 NameServer处理主流程代码
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
if (log.isDebugEnabled()) {
log.debug("receive request, {} {} {}",
request.getCode(),
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
request);
}
switch (request.getCode()) {
case RequestCode.PUT_KV_CONFIG:
return this.putKVConfig(ctx, request);
case RequestCode.GET_KV_CONFIG:
return this.getKVConfig(ctx, request);
case RequestCode.DELETE_KV_CONFIG:
return this.deleteKVConfig(ctx, request);
case RequestCode.REGISTER_BROKER:
Version brokerVersion = MQVersion.value2Version(request.getVersion());
if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
return this.registerBrokerWithFilterServer(ctx, request);
} else {
return this.registerBroker(ctx, request);
}
case RequestCode.UNREGISTER_BROKER:
return this.unregisterBroker(ctx, request);
case RequestCode.GET_ROUTEINTO_BY_TOPIC:
return this.getRouteInfoByTopic(ctx, request);
case RequestCode.GET_BROKER_CLUSTER_INFO:
return this.getBrokerClusterInfo(ctx, request);
case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
return this.wipeWritePermOfBroker(ctx, request);
case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
return getAllTopicListFromNameserver(ctx, request);
case RequestCode.DELETE_TOPIC_IN_NAMESRV:
return deleteTopicInNamesrv(ctx, request);
case RequestCode.GET_KVLIST_BY_NAMESPACE:
return this.getKVListByNamespace(ctx, request);
case RequestCode.GET_TOPICS_BY_CLUSTER:
return this.getTopicsByCluster(ctx, request);
case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
return this.getSystemTopicListFromNs(ctx, request);
case RequestCode.GET_UNIT_TOPIC_LIST:
return this.getUnitTopicList(ctx, request);
case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
return this.getHasUnitSubTopicList(ctx, request);
case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
return this.getHasUnitSubUnUnitTopicList(ctx, request);
case RequestCode.UPDATE_NAMESRV_CONFIG:
return this.updateConfig(ctx, request);
case RequestCode.GET_NAMESRV_CONFIG:
return this.getConfig(ctx, request);
default:
break;
}
return null;
}
在Consumer的源码中，获取消息的底层通信部分同样发送一个RemotingCommand请求，返回的response也是个RemotingCommand类型，如代码清单4-8所示。
代码清单4-8 Consumer请求消息底层实现代码
private PullResult pullMessageSync(//
final String addr, // 1
final RemotingCommand request, // 2
final long timeoutMillis// 3
) throws RemotingException, InterruptedException, MQBrokerException {
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
assert response != null;
return this.processPullResponse(response);
}
从源码中可以看出，RocketMQ中复杂的通信过程，被RemotingCommand统一起来，大部分的逻辑都是通过发送、接受并处理Command来完成的。
2.协议设计和编解码
RocketMQ自己定义了一个通信协议，使得模块间传输的二进制消息和有意义的内容之间互相转换。协议格式如图4-2所示。
图4-2 RocketMQ的通信协议
1）第一部分是大端4个字节整数，值等于第二、三、四部分长度的总和；
2）第二部分是大端4个字节整数，值等于第三部分的长度；
3）第三部分是通过Json序列化的数据；
4）第四部分是通过应用自定义二进制序列化的数据。
消息的解码过程在RemotingCommand的decode函数里，如代码清单4-9所示。
代码清单4-9 消息解码函数
ublic static RemotingCommand decode(final ByteBuffer byteBuffer) {
int length = byteBuffer.limit();
int oriHeaderLen = byteBuffer.getInt();
int headerLength = getHeaderLength(oriHeaderLen);
byte[] headerData = new byte[headerLength];
byteBuffer.get(headerData);
RemotingCommand cmd = headerDecode(headerData, getProtocolType (oriHeaderLen));
int bodyLength = length - 4 - headerLength;
byte[] bodyData = null;
if (bodyLength > 0) {
bodyData = new byte[bodyLength];
byteBuffer.get(bodyData);
}
cmd.body = bodyData;
return cmd;
}
对应的消息编码过程在RemotingCommand的encode函数中，如代码清单4-10所示。
代码清单4-10 消息编码函数
public ByteBuffer encode() {
// 1> header length size
int length = 4;
// 2> header data length
byte[] headerData = this.headerEncode();
length += headerData.length;
// 3> body data length
if (this.body != null) {
length += body.length;
}
ByteBuffer result = ByteBuffer.allocate(4 + length);
// length
result.putInt(length);
// header length
result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
// header data
result.put(headerData);
// body data;
if (this.body != null) {
result.put(this.body);
}
result.flip();
return result;
}
3.Netty库
RocketMQ是基于Netty库来完成RemotingServer和RemotingClient具体的通信实现的，Netty是个事件驱动的网络编程框架，它屏蔽了Java Socket、NIO等复杂细节，用户只需用好Netty，就可以实现一个“网络编程专家+并发编程专家”水平的Server、Client网络程序。应用Netty有一定的门槛，需要了解它的EventLoopGroup、Channel、Handler模型以及各种具体的配置。RocketMQ利用Netty实现的通信类是NettyRemotingServer和NettyRemotingClient，用户也可以参考这两个类的实现来学习使用Netty。
