RocketMQ底层通信机制
分布式系统各个角色间的通信效率很关键，通信效率的高低直接影响系统性能，基于Socket实现一个高效的Tcp通信协议是个很有挑战的事情，本节说明RocketMQ是如何解决这个问题的 1.1.1 Remoting模块RocketMQ的通信相关代码在Remoting模块里，先来看看主要类结构。 RemotingService为最上层接口,定义了三个方法：void start();void shutdown();void registerRPCHook(RPCHook rpcHook);RemotingClient，RemotingServer继承RemotingService接口, 并增加了自己特有的方法。代码清单1-1 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 addrs); 然后看看具体的实现类，NettyRemotingClient和NettyRemotingServer分别实现了RemotingClient和RemotingServer, 而且都继承了NettyRemotingAbstract类. 通过上面的封装，RocketMQ各个模块间的通信，可以通过发送统一格式的自定义消息(RemotingCommand)来完成的，各个模块间的通信实现简洁明了。比如NameServer模块中，NameServerController有个remotingServer变量，NameServer在启动时初始化好各个变量，然后启动remotingServer即可，剩下NameServer要做的是专心实现好处理RemotingCommand的逻辑。代码清单1-2 NameServer处理主流程代码@Overridepublic 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.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的源码中，获取消息的底层的通信部分也是发送一个RemotingComand 请求，返回的response也是个RemotingCommand类型。代码清单1-3 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，接受并处理Command完成。1.1.2 协议设计和编解码RocketMQ自己定义了一个通信协议，使得模块间传输的二进制消息和有意义的内容之间互相转换。协议格式如图4-2所示。 图1-2 RocketMQ的通信协议 （1）第一部分是大端4个字节整数，值等于第二，三，四部分长度总和 （2）第二部分是大端4个字节整数，值等于第三部分的长度 （3）第三部分是通过json 序列化的数据 （4）第四部分是通过应用自定义二进制序列化的数据 消息的解码过程在RomotingCommand的decode函数里。 代码清单1-4 消息解码函数 public 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函数中。代码清单1-5 消息编码函数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; } 1.1.3 Netty库RocketMQ是基于Netty库来完成RemotingServer和RemotingClient具体的通信实现的，Netty是个事件驱动的网络编程框架，它屏蔽了Java Socket，Nio等复杂细节，用户只需用好Netty，就可以实现一个网络编程专家+并发编程专家水平的Server、Client网络程序。应用Netty有一定的门槛，需要了解它的EventLoopGroup，Channel，Handler模型以及各种具体的配置。RocketMQ利用Netty实现的通信类是NettyRemotingServer和NettyRemotingClient，用户也可以参考这两个类的实现来学习使用Netty。 推荐阅读： 云栖社区官方出品 RocketMQ实战与原理解析作者：杨开元定价：59.00元•RocketMQ由阿里开源，Apache开源项目，经受多年流量峰值考验，在多个性能指标上远超同类产品•作者是阿里资深数据专家，有多年RocketMQ使用经验，深入研究RocketMQ源代码，写作前与RocketMQ官方团队有深入沟通•云栖社区官方出品，得到RocketMQ官方研发团队以及业界的多位专家的肯定和推荐 阅读原文：http://product.dangdang.com/25290633.html
