RocketMQ的底层消息存储架构以及优化措施

RocketMQ的底层消息存储架构以及优化措施

详细介绍了RocketMQ的消息存储架构，以及效率优化机制，Mmap内存映射以及Page Cache页缓存机制。

文章目录

• 1 消息存储架构
• 2 混合型的存储结构
• 3 页缓存与内存映射
• 4 内存预映射和文件预热机制

1 消息存储架构

RocketMQ 消息存储架构图如下：

消息存储架构图中主要有下面三个跟消息存储相关的文件构成：

1. CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容，消息内容不是定长的。单个文件大小默认1G ，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息顺序写入日志文件，效率很高，当文件满了，写入下一个文件。
2. ConsumeQueue：消息消费队列（可以理解为Topic中的队列），引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值，以及ConsumeOffset（每个消费者组的消费位置）。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。
   1. 所以说，一个queueId所表示的队列实际只是一个文件目录，其内部可能有多个ConsumeQueue文件，这些文件共同组成了某个topic下的queueId所表示的逻辑队列。
3. IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$HOME/store/index${fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

2 混合型的存储结构

在上面的RocketMQ的消息存储整体架构图中可以看出，RocketMQ采用的是混合型的存储结构，即为单个Broker实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储，或者说多个Topic的消息实体内容都存储于一个CommitLog文件中。与RocketMQ不同的是，在Kafka 中则会为每个 Partition分配一个单独的存储文件。

RocketMQ的混合型存储结构针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。

正因为如此，Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后，可以等下一次消息拉取，同时服务端也支持长轮询模式，如果一个消息拉取请求未拉取到消息，Broker允许等待15s的时间，只要这段时间内有新消息到达，将直接返回给消费端。这里，RocketMQ的具体做法是，使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。

Consumer端先从ConsumeQueue（消息逻辑队列）这个二级索引队列读取持久化消息的起始物理位置偏移量offset、大小size和消息Tag的HashCode值，随后再从CommitLog中进行读取待拉取消费消息的真正实体内容部分。

RocketMQ的混合型存储结构，ConsumeQueue消息逻辑队列存储的数据较少，并且对磁盘的是顺序读取，在PageCache机制的预读取作用下，ConsumeQueue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。

3 页缓存与内存映射

页缓存（PageCache)是OS对文件的缓存（属于内核空间的内存），用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问 *** 作进行了性能优化，将一部分的内存用作PageCache。对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。

在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在PageCache机制的预读取作用下，ConsumeQueue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Deadline”（此时块存储采用SSD的话），随机读的性能也会有所提升。

另外，RocketMQ主要通过Mmap内存映射技术（Memory Mapped Files）对文件进行读写 *** 作（MappedByteBuffer类）。其中，首先利用了NIO中的FileChannel模通过map()方法（本质是对系统调用mmap()的封装）将磁盘上的物理文件直接映射到用户态的内存地址中（获取MappedByteBuffer），将对文件的 *** 作转化为直接对内存地址进行 *** 作，从而极大地提高了文件的读写效率。因为这种Mmap的方式减少了传统IO将磁盘文件数据在 *** 作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销。而正因为需要使用内存映射机制，故RocketMQ的文件存储都使用定长结构来存储，方便一次将整个文件映射至内存。

简单的说，Mmap技术和普通write/read一样需要从pagecache中进行读写，但是mmap系统调用读写数据会直接到达内核空间中的pagecache（应用程序可以直接访问这一块内核空间），减少了数据在用户态内核态之间的一次复制，但是并没有减少上下文切换次数。

RocketMQ通过mmap技术减少数据拷贝次数，然后利用pagecache技术实现尽可能优先读写内存，实现消息的快速读写。

Kakfa对于消息的写入也是采用的Mmap技术，但在消息的消费时则是使用的zero-copy技术（sendfile系统调用，零拷贝即没有CPU拷贝，仅有DMA拷贝），因此，在发送消息的时候基于sendfile的Zero-Copy技术相比于RocketMQ的Mmap+write性能更好一些，减少了上下文切换次数和数据拷贝次数。

某些文章中，mmap和sendfile都算作Zero-Copy技术的实现。

4 内存预映射和文件预热机制

如果要写消息到CiommitLog文件，首先会通过mmap将文件通过MappedByteBuffer映射的内存中，然后通过写MappedByteBuffer，将数据直接写入PageCache中，然后由os的线程定时异步刷入磁盘中。因为大部分情况下，消息到达Broker之后会很快被消费，所以在读取的时候，消费者也可以直接从PageCache读取的数据，因此，总体情况下，消息的读写都是读写的内存，磁盘中的数据会被滞后的更新。

mmap *** 作需要在映射的时候确定内存大小，并且最大不得超过1.5G，因此CiommitLog文件默认固定1G，并且在进行映射的时候也会需要时间。因此，RocketMQ使用内存预映射机制，来提前对一些可能接下来要读写的磁盘文件进行mmap *** 作，这样后续读写文件的时候，就不需要再执行mmap *** 作了。

但是，调用Mmap进行内存映射后，OS仅仅是分配了内存并建立虚拟内存地址至物理地址的映射表，实际并没有加载任何文件内容至内存的PageCache中，默认情况下，只有第一次访问的时候OS才会检查该部分数据是否已经在内存中，如果不在，则发出一次缺页中断，从而进行磁盘IO将数据加载到PageCache中来，所以，仍然有可能会在mmap之后频繁的从磁盘里加载数据到内存中去。为此，RocketMQ在执行mmap调用之后执行madvise系统调用，尽可能多的把磁盘文件加载到内存里去。

到此，我们知道RocketMQ除了mmap和PageCache机制，在内部还通过内存预映射和文件预热机制，把磁盘文件里的数据尽可能多的加载到PageCache也就是内存中来，进一步提升了读写的效率。

相关文章：

RocketMQ

如有需要交流，或者文章有误，请直接留言。另外希望点赞、收藏、关注，我将不间断更新各种Java学习博客！