RocketMQ-消息存储

待我称王封你为后i 2022-12-06 12:43 239阅读 0赞

# 一、磁盘文件结构 #

## 1.1文件简介 ##

RocketMQ的Broker机器磁盘上的文件存储结构  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70]

1.  CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容。单个文件大小默认1G，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；
2.  ConsumeQueue：消息消费的逻辑队列，作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。而IndexFile（索引文件）则只是为了消息查询提供了一种通过key或时间区间来查询消息的方法（ps：这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程）。从实际物理存储来说，ConsumeQueue对应每个Topic和QueuId下面的文件。单个文件大小约5.72M，每个文件由30W条数据组成，每个文件默认大小为600万个字节，当一个ConsumeQueue类型的文件写满了，则写入下一个文件；
3.  IndexFile：因为所有的消息都存在CommitLog中，如果要实现根据 key 查询消息的方法，就会变得非常困难，所以为了解决这种业务需求，有了IndexFile的存在。用于为生成的索引文件提供访问服务，通过消息Key值查询消息真正的实体内容。在实际的物理存储上，文件名则是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引；

## 1.2文件格式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 1]

### 1.2.1CommitLog ###

消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容。消息存放的物理文件，每台broker上的commitlog被本机所有的queue共享，不做任何区分。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center]

### 1.2.2ConsumeQueue ###

消息消费队列，消息到达 CommitLog 文件后，将异步转发到消息 消费队列，供消息消费者消费。ConsumeQueue存储格式如下：  
![在这里插入图片描述][20200913113152639.png]  
单个 ConsumeQueue 文件中默认包含 30 万个条目，单个文件的长度为 30w × 20 字节， 单个 ConsumeQueue 文件可以看出是一个 ConsumeQueue 条目的数组，其下标为 ConsumeQueue 的逻辑偏移量，消息消费进度存储的偏移量 即逻辑偏移量。  
ConsumeQueue 即为 Commitlog 文件的索引文件， 其构建机制是当消息到达 Commitlog 文件后， 由专门的线程 产生消息转发任务，从而构建消息消费队列文件与下文提到的索引文件。

### 1.2.3IndexFile ###

消息索引文件，如果一个消息包含key值的话，会使用IndexFile存储消息索引，主要存储消息 Key 与 Offset 的对应关系。消息消费队列是RocketMQ专门为消息订阅构建的索引文件，提高根据主题与消息队 列检索消息的速度 ，另外 RocketMQ 引入了 Hash 索引机制为消息建立索引， HashMap 的设 计包含两个基本点 ： Hash 槽与 Hash 冲突的链表结构。 RocketMQ 索引文件布局如图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 3]  
lndexFile 总共包含 lndexHeader、 Hash 槽、 Hash 条目，索引文件主要用于根据key来查询消息的，流程主要是：

1、根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目，例如图中所示 slotNum=5000000)
    2、根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)
    3、遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)

# 二、消息读写流程 #

## 2.1消息存储整体架构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center 1]  
上图即为RocketMQ的消息存储整体架构，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ采用混合型存储结构的缺点在于，会存在较多的随机读操作，因此读的效率偏低。同时消费消息需要依赖ConsumeQueue，构建该逻辑消费队列需要一定开销。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center 2]

从上面的整体架构图中可见，RocketMQ的混合型存储结构针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。正因为如此，Consumer也就肯定有机会去消费这条消息，至于消费的时间可以稍微滞后一些也没有太大的关系。退一步地讲，即使Consumer端第一次没法拉取到待消费的消息，Broker服务端也能够通过长轮询机制等待一定时间延迟后再次发起拉取消息的请求。  
Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue和IndexFile数据。然后，Consumer即可根据ConsumerQueue来查找待消费的消息了。其中，ConsumeQueue作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。而IndexFile则只是为了消息查询提供了一种通过key或时间区间来查询消息的方法。

## 2.2发送消息 ##

发送时，Producer不直接与Consume Queue打交道。上文提到过，RMQ所有的消息都会存放在Commit Log中，为了使消息存储不发生混乱，多线程对Commit Log写会上锁。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 4]

图片：Commit Log顺序写  
消息持久被锁串行化后，对Commit Log就是顺序写，也就是常说的Append操作。配合上Page Cache，RMQ在写Commit Log时效率会非常高。正因为写Commit Log很快，RMQ也大胆提供了自旋锁以提高性能。  
Commit Log持久后，Broker会将消息逐个异步Dispatch到对应的Consume Queue文件中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 5]  
图片：Consume Queue顺序写  
每一个Consume Queue代表一个逻辑队列，是由ReputMessageService在单个Thread Loop中Append，如上图所示，每一个Consume Queue显然也是从左往右高效顺序写。

## 2.3消费消息 ##

消费时，Consumer不直接与Commit Log打交道，而是从Consume Queue中去拉取数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 6]  
图片：Consume Queue顺序读  
如上图所示，拉取的顺序从旧到新，每一个Consume Queue都是顺序读。  
光拉取Consume Queue是没有消息真实内容的，但是里面有对Commit Log的偏移值引用，所以再次映射到Commit Log获取真实消息数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 7]  
图片：Commit Log随机读  
问题出现了，从上图可以看到，Commit Log会进行随机读。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 8]  
图片：Commit Log整体有序随机读  
虽然是随机读，但整体还是从旧到新有序读，只要随机的那块区域还在Page Cache的热点范围内，还是可以充分利用Page Cache。

通常文件随机读写非常慢，但对文件进行顺序读写，速度几乎是接近于内存的随机读写，为什么会这么快，原因就是OS对文件IO有优化。OS发现系统的物理内存有大量剩余时，为了提高IO的性能，将一部分的内存用作Page Cache。

OS在读磁盘时会按照文件顺序预先将内容读到Cache中，以便下次读时能命中Cache，写磁盘时直接写到Cache中就写返回，由pdflush以某种策略将Cache的数据Flush回磁盘。

文件顺序IO时，读和写的区域都是被OS智能Cache过的热点区域，不会产生大量缺页（Page Fault）中断而再次读取磁盘，文件的IO几乎等同于内存的IO。

发送消息时，消息要写进Page Cache而不是直接写磁盘，依赖异步线程刷盘；接收消息时，消息从Page Cache直接获取而不是缺页从磁盘读取，并且Cache本身就由内核管理，不需要从程序到内核的数据Copy，直接通过Socket传输。

## 2.4PageCache与Mmap内存映射 ##

有必要简单地介绍下page cache的概念。系统的所有文件I/O请求，操作系统都是通过page cache机制实现的。对于操作系统来说，磁盘文件都是由一系列的数据块顺序组成，数据块的大小由操作系统本身而决定，x86的linux中一个标准页面大小是4KB。

操作系统内核在处理文件I/O请求时，首先到page cache中查找（page cache中的每一个数据块都设置了文件以及偏移量地址信息），如果未命中，则启动磁盘I/O，将磁盘文件中的数据块加载到page cache中的一个空闲块，然后再copy到用户缓冲区中。

page cache本身也会对数据文件进行预读取，对于每个文件的第一个读请求操作，系统在读入所请求页面的同时会读入紧随其后的少数几个页面。因此，想要提高page cache的命中率（尽量让访问的页在物理内存中），从硬件的角度来说肯定是物理内存越大越好。从操作系统层面来说，访问page cache时，即使只访问1k的消息，系统也会提前预读取更多的数据，在下次读取消息时, 就很可能可以命中内存。

在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在page cache机制的预读取作用下，Consume Queue的读性能会比较高近乎内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Noop”（此时块存储采用SSD的话），随机读的性能也会有所提升。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center 3]  
上图中，整个OS有3.7G的物理内存，用掉了2.7G，应当还剩下1G空闲的内存，但OS给出的却是175M。因为OS发现系统的物理内存有大量剩余时，为了提高IO的性能，就会使用多余的内存当做文件缓存，也就是图上的buff/cache，广义我们说的Page Cache就是这些内存的子集。

另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型直接将磁盘上的物理文件直接映射到用户态的内存地址中（这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销），将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了）。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 9]

参考：  
[RocketMQ相关流程图/原理图][RocketMQ]  
[RocketMQ之六：RocketMQ消息存储][RocketMQ_RocketMQ]  
[RocketMQ高性能之底层存储设计][RocketMQ 1]  
[消息中间件—RocketMQ消息存储（一）][RocketMQ 2]  
[RocketMQ消息存储][RocketMQ 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70]: /images/20221123/2e0a622304b64c54ba99afa9e3c8eed6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 1]: /images/20221123/912fd72401094aed9704cb891e463a63.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 2]: /images/20221123/8b6c64ce603f41c4ad6dec2592ab71de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/93680a7e77e0454682f0be798efa924d.png
[20200913113152639.png]: /images/20221123/5ceb9d51dfa64cb68372bfafeb3f0280.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 3]: /images/20221123/441f7ffe020748fba1e9e674e67df647.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/c7b9739b00e54f0e9c049814027d6376.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/9c7e19f6f9ca46d6a921d2aec87b0cb4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 4]: /images/20221123/f90d5945ca154db78a504794727f4732.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 5]: /images/20221123/0056a67ab59246829341cfcd936bfecb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 6]: /images/20221123/707ef435825a41adb6a3ea620af520cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 7]: /images/20221123/0a854971f32a4bfcbe7de8ffa122ded7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 8]: /images/20221123/5425cc6ae1ef42b89660031c130a84af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/d282081cec4d43269dc2c55e2f1a5c97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpaHVheW9uZw_size_16_color_FFFFFF_t_70 9]: /images/20221123/bd720f8ae1e343c395f4f2f14328bde0.png
[RocketMQ]: https://juejin.im/post/6844903941839437838
[RocketMQ_RocketMQ]: https://www.cnblogs.com/duanxz/p/5020398.html
[RocketMQ 1]: https://mp.weixin.qq.com/s/yd1oQefnvrG1LLIoes8QAg
[RocketMQ 2]: https://www.jianshu.com/p/b73fdd893f98
[RocketMQ 3]: https://www.cnblogs.com/fanguangdexiaoyuer/p/10496112.html