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处理主流程代码
@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.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)第一部分是大端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官方研发团队以及业界的多位专家的肯定和推荐
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
学习嵌入式,我们到底该学些什么?
嵌入式越来越复杂,一个SOC芯片上集成的模块越来越多。以手机为例,典型的嵌入式产品,我们看看上面集成了多少模块:触摸屏、LCD、USB、WiFi、4G等无线通信、音视频编解码IP、DDR、存储控制器、3D/2D加速、GPS、指纹识别、NFC、DMA、G-sensor各种传感器.......。可以说,现在一个手机的复杂度和硬件配置,已经超过我们的桌面PC了。除了不断增加的硬件,软件方面,比如Linux内核,光内核代码就有1000多万行,每天更新的速度超过你学习的进度,你能学得完嵌入式的所有知识和技能吗?工业主板 嵌入式系统开发 早期PC时代,我们知道能做出X86 CPU量产的也没有几家,Intel、AMD和威盛。但是嵌入式时代不一样了,ARM的IP授权模式导致不同的芯片厂商百家齐放,不同的SOC平台和开发板眼花缭乱,针对不同行业需求定制的SOC平台雨后春笋:手机芯片、平板芯片、视频安防、物联网、汽车电子、工业控制,甚至人工智能AI芯片....,你到Linux内核的ARCH下面可以看看有多少种CPU架构,再到arch/arm下面看看有多少种开发平台,这还只是加入到内核mainline的平台...
- 下一篇
spring cloud(学习笔记) Enreka服务治理
服务治理是微服务架构最为核心和基础的模块,主要用来实现各个微服务实例的自动化注册和发现。 记录一下服务注册中心的搭建以及高可用注册中心的实现 1.首先创建两个基础 的spring boot工程,spring boot创建工程的网站:http://start.spring.io/,创建界面如下 2.解压工程,用Maven的形式导入工程(File->new->project from Existing Soures,选择解压工程导入) 3.两个工程中都添加依赖(eureka) 1 <properties> 2 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> 3 <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> 4 <java.version>1.8</java.version> 5 <spring-clou...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS7,CentOS8安装Elasticsearch6.8.6
- CentOS6,CentOS7官方镜像安装Oracle11G
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- CentOS8编译安装MySQL8.0.19
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- Windows10,CentOS7,CentOS8安装Nodejs环境
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- CentOS8,CentOS7,CentOS6编译安装Redis5.0.7