RocketMQ
RocketMQ是一个统一的消息传递引擎,轻量级的数据处理平台。
Name Server
Name Server充当路由消息的提供者,生产者(Producer)或消费者(Customer)可以通过Name Server查找各主题对应的Broker IP列表,多个Name Server实例组成集群,但相互独立,没有信息交换。支持Broker的动态注册与发现。主要包括两个功能:
Broker管理:Name Server接收Broker集群的注册信息并且保存下来作为路由信息的基本数据,然后提供心跳检测机制,检查Broker是否存活;
路由信息管理:每个Name Server将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息,然后生产者(Producer)和消费者(Conumser)通过Name Server就可以知道整个Broker集群的整个路由信息和用于客户端查询的队列信息,从而进行消息的投递和消费。
Broker Server
负责存储消息、转发消息。Broker Server在RocketMQ系统中负责接收从生产者(Producer)发送来的消息并存储、同事为消费者的拉取请求做准备。Borker Server也存储消息相关的元数据,包括消费者组、消费进度偏移、主题和队列信息等。
Broker启动后需要完成一次将自己注册至NameServer的操作;随后每隔30s时间定时向NameServer上报Topic路由信息。
为了实现Broker中消息的存储、投递、查询以及服务高可用的保证,囊括一下几个重要的子模块信息:
Remoting Module:整个borker的实体,负责处理来自客户端的请求;
Client Manager:负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息;
Store Service:提供简单的API接口处理消息存储到物理硬盘和查询功能;
HA Service:高可用服务,提供Master Broker和Slaver Broker之间的数据同步功能;
Index Service:根据特定的Message key对投递到Broker的消息进行索引服务,以提供消息的快速查询。
Broker 架构
消息存储是RocketMQ中至关重要的一部分,想对其他来说会比较复杂。
消息存储架构
- CommitLog:消息主体以及元数据的存储主体,存储Producer端写入的消息主体内容,消息内容不是定长的,单个文件大小默认1G,文件名长度20位,左边补0,剩余为其实偏移量,比如00000000000000000000代表了第一个文件,其实偏移量为0,文件大小为1G=1073741824;当第一个文件写满了,第二个文件为00000000001073741824,起始偏移量为1073741824,以此类推,消息主要是顺序写入日志文件,当文件满了,写入下一个文件;
- ConsumeQueue:消息消费队列,引入的目的主要是提高消费消息的性能,由于RocketMQ是基于主题topic的订阅模式,消息消费是针对主题进行的,如果要遍历commitlog文件中根据topic检索消息非常低效的,Consumer即可根据ConsumeQueue来查找待消费的消息,其中,ConsumeQueue(逻辑消费队列)作为消费消息的索引,保存指定Topic下的队列消息在CommitLog中的起始物理偏移量offset,消息大小size和消息Tag的hashCode值,consumequeue文件可以看成是基于topic的commitLog索引文件,所以consumequeue文件夹组织方式如下:topic/queue/file三层组织结构,具体存储路径为:$HOME/store/consumequeue/{topic}/{queueId}/{fileName},同样consumequeue文件采用定长设计,每一个条目共20个字节,分别为8字节的commitlog物理偏移量,4字节消息长度,8字节tag hashcode,单个文件由30W个条目组成,可以像数组一样随机访问每一个条目,每个ConsumeQueue文件大小约为5.72M;
- IndexFile:IndexFile(索引文件)提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是:$HOME/store/index/{fileName},文件名fileName是以创建时的时间戳命名的,固定的单个IndexFile文件大小约为400M,一个IndexFile可以保存 2000W个索引,IndexFile的底层存储设计为在文件系统中实现HashMap结构,故rocketmq的索引文件其底层实现为hash索引。
消息刷盘
- 同步刷盘:只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障,但是性能上会有较大影响,一般适用于金融业务应用该模式较多。
- 异步刷盘:能够充分利用OS的PageCache的优势,只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行,降低了读写延迟,提高了MQ的性能和吞吐量。
Broker 角色
Broker角色分为ASYNC_MASTER(异步主机)、SYNC_MASTER(同步主机)以及SLAVE(从机)。
SYNC_MASTER(同步主机):对信息的可靠性要求比较严格;
ASYNC_MASTER(异步主机):对消息可靠性要求不高;
Broker 配置
参数名 | 默认值 | 说明 |
---|---|---|
listenPort | 10911 | 接受客户端连接的监听端口 |
namesrvAddr | Null | nameServer地址 |
brokerIP1 | 网卡的InetAddress | 当前broker监听的IP |
brokerIP2 | 跟brokerIP1一样 | 存在主从broker时,如果在broker主节点上配置了brokerIP2属性,broker从节点会连接主节点配置的brokerIP2进行同步 |
brokerName | null | broker的名称 |
brokerClusterName | DefaultCluster | broker的名称 |
brokerId | 0 | 0表示master,其他的正整数表示slave |
storePathCommitLog | $HOME/store/commitlog/ | 存储 commit log 的路径 |
storePathConsumerQueue | $HOME/store/consumequeue/ | 存储 consume queue 的路径 |
mappedFileSizeCommitLog | 1024 * 1024 * 1024(1G) | commit log 的映射文件大小 |
deleteWhen | 04 | 在每天的什么时间删除已经超过文件保留时间的 commit log |
fileReservedTime | 72 | 以小时计算的文件保留时间 |
brokerRole | ASYNC_MASTER | SYNC_MASTER/ASYNC_MASTER/SLAVE |
flushDiskType | ASYNC_FLUSH | SYNC_FLUSH/ASYNC_FLUSH SYNC_FLUSH 模式下的 broker 保证在收到确认生产者之前将消息刷盘。ASYNC_FLUSH 模式下的 broker 则利用刷盘一组消息的模式,可以取得更好的性能。 |
nameSrv和Broker简单的架构图
</article>