13.RocketMQ之消息的存储与发送-Toy模板网

这篇具有很好参考价值的文章主要介绍了13.RocketMQ之消息的存储与发送。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1. 消息存储

1.1 消息存储

分布式队列因为有高可靠性的要求，所以数据要进行持久化存储。

13.RocketMQ之消息的存储与发送

消息生成者发送消息
Broker收到消息，将消息进行持久化，在存储中新增一条记录
返回ACK给生产者
Broker消息给对应的消费者，然后等待消费者返回ACK
如果消息消费者在指定时间内成功返回ack，那么MQ认为消息消费成功，在存储中删除消息，即执行第6步；如果MQ在指定时间内没有收到ACK，则认为消息消费失败，会尝试重新push消息,重复执行4、5、6步骤
MQ删除消息

1.1.1 存储介质

关系型数据库DB

Apache下开源的另外一款MQ—ActiveMQ(默认采用的KahaDB做消息存储)可选用JDBC的方式来做消息持久化，通过简单的xml配置信息即可实现JDBC消息存储。由于，普通关系型数据库(如Mysql)在单表数据量达到千万级别的情况下，其IO读写性能往往会出现瓶颈。在可靠性方面，该种方案非常依赖DB，如果一旦DB出现故障，则MQ的消息就无法落盘存储会导致线上故障

文件系统

目前业界较为常用的几款产品(RocketMQ/Kafka/RabbitMQ)均采用的是消息刷盘至所部署虚拟机/物理机的文件系统来做持久化(刷盘一般可以分为异步刷盘和同步刷盘两种模式)。

消息刷盘为消息存储提供了一种高效率、高可靠性和高性能的数据持久化方式。

除非部署MQ机器本身或是本地磁盘挂了，否则一般是不会出现无法持久化的故障问题。

1.1.2 性能对比

文件系统>关系型数据库DB

1.1.3 消息的存储和发送

1.1.3.1消息存储:顺序写

磁盘如果使用得当，磁盘的速度完全可以匹配上网络的数据传输速度。目前的高性能磁盘，顺序写速度可以达到600MB/s，超过了一般网卡的传输速度。但是磁盘随机写的速度只有大概100KB/s，和顺序写的性能相差6000倍！因为有如此巨大的速度差别，好的消息队列系统会比普通的消息队列系统速度快多个数量级。RocketMQ的消息用顺序写,保证了消息存储的速度。

1.1.3.2消息发送

Linux操作系统分为【用户态】和【内核态】，文件操作、网络操作需要涉及这两种形态的切换，免不了进行数据复制。

一台服务器把本机磁盘文件的内容发送到客户端，一般分为两个步骤：

1)read；读取本地文件内容；

2)write；将读取的内容通过网络发送出去。

这两个看似简单的操作，实际进行了4 次数据复制，分别是：

从磁盘复制数据到内核态内存；
从内核态内存复制到运行程序用户态内存；
然后从运行程序用户态内存复制到网络驱动的内核态内存；
最后是从网络驱动的内核态内存复制到网卡中进行传输。

13.RocketMQ之消息的存储与发送

通过使用mmap的方式，可以省去向用户态的内存复制，提高速度。这种机制在Java中是通过MappedByteBuffer实现.

13.RocketMQ之消息的存储与发送

RocketMQ充分利用了上述特性，也就是所谓的“零拷贝”技术，提高消息存盘和网络发送的速度。

这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5\~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了

1.1.4 消息存储结构

RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。

每个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。

13.RocketMQ之消息的存储与发送

CommitLog：存储消息的元数据
ConsumerQueue：消息在CommitLog的索引。记录了minOffset和maxOffset和已经消费了的消息的偏移量consumerOffset。可以通过offset去CommitLog查找消息。
IndexFile：为了消息查询提供了一种通过key或时间区间来查询消息的方法，这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程。
ConsumerQueue和messageQueue是一一对应的。