在Client和Server之间完成一次消息发送时,需要对发送的消息进行一个协议约定,因此就有必要自定义RocketMQ的消息协议;为了高效地在网络中传输消息和对收到的消息读取,就需要对消息进行编解码,在RocketMQ中,RemotingCommand这个类在消息传输过程中对所有数据内容的封装,不但包含了所有的数据结构,还包含了编码解码操作;
{
"code": 0,
"language": "JAVA",
"version": 0,
"opaque": 0,
"flag": 1,
"remark": "hello, I am respponse /127.0.0.1:27603",
"extFields": {
"count": "0",
"messageTitle": "HelloMessageTitle"
}
}
在RocketMQ消息队列中支持通信的方式主要有同步(sync)、异步(async)、单向(oneway) 三种,其中“单向”通信模式相对简单,一般用在发送心跳包场景下,无需关注其Response;
RocketMQ的RPC通信采用Netty组件作为底层通信库,同样也遵循了Reactor多线程模型,同时又在这之上做了一些扩展和优化;
一个Reactor主线程(eventLoopGroupBoss,即为上面的1)负责监听TCP网络连接请求,建立好连接,创建SocketChannel,并注册到selector上; RocketMQ的源码中会自动根据OS的类型选择NIO和Epoll,也可以通过参数配置),然后监听真正的网络数据。拿到网络数据后,再丢给Worker线程池(eventLoopGroupSelector,即为上面的“N”,源码中默认设置为3),在真正执行业务逻辑之前需要进行SSL验证、编解码、空闲检查、网络连接管理,这些工作交给defaultEventExecutorGroup(即为上面的“M1”,源码中默认设置为8)去做;处理业务操作放在业务线程池中执行,根据RomotingCommand的业务请求码code去processorTable这个本地缓存变量中找到对应的 processor,然后封装成task任务后,提交给对应的业务processor处理线程池来执行(sendMessageExecutor,以发送消息为例,即为上面的“M2”);从入口到业务逻辑的几个步骤中线程池一直再增加,这跟每一步逻辑复杂性相关,越复杂,需要的并发通道越宽;