RocketMQ因为有高可靠性的要求(宕机不丢失数据)，所以数据要进行持久化存储。所以RocketMQ采用文件进行存储。

物理文件结构

RocketMQ的数据默认存储在${user.home}/store目录下，可以通过修改broker.conf中的参数storePathRootDir的值进行设置。

物理目录结构大致如下：

├── abort
├── checkpoint
├── commitlog
│   ├── 00000000000000000000
│   └── 00000000001073741824
├── config
│   ├── consumerFilter.json
│   ├── consumerFilter.json.bak
│   ├── consumerOffset.json
│   ├── consumerOffset.json.bak
│   ├── delayOffset.json
│   ├── delayOffset.json.bak
│   ├── subscriptionGroup.json
│   ├── topics.json
│   └── topics.json.bak
├── consumequeue
│   ├── TopicTest
│   │   ├── 0
│   │   │   └── 00000000000000000000
│   │   ├── 1
│   │   │   └── 00000000000000000000
│   │   ├── 2
│   │   │   └── 00000000000000000000
│   │   └── 3
│   │       └── 00000000000000000000
│   └── TopicTest1
│       ├── 0
│       │   └── 00000000000000000000
│       └── 1
│           └── 00000000000000000000
├── index
│   └── 20220706092335158
└── lock

目录结构说明：

• commitLog：消息存储目录
• config：运行期间一些配置信息
• consumerqueue：消息消费队列存储目录
• index：消息索引文件存储目录
• abort：如果存在改文件则Broker非正常关闭
• checkpoint：文件检查点，存储CommitLog文件最后一次刷盘时间戳、consumerqueue最后一次刷盘时间，index索引文件最后一次刷盘时间戳。

逻辑结构

RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。

• CommitLog：存储消息的元数据
• ConsumerQueue：存储消息在CommitLog的索引
• IndexFile：为了消息查询提供了一种通过key或时间区间来查询消息的方法，这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程

CommitLog

CommitLog以物理文件的方式存放，每台Broker上的CommitLog被本机器所有ConsumeQueue共享。在CommitLog中，一个消息的存储长度是不固定的， RocketMQ采取一些机制，尽量向CommitLog中顺序写，但是随机读。commitlog文件默认大小为lG ，可通过在broker配置文件中设置mappedFileSizeCommitLog属性来改变默认大小。

CommitLog作为消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容，消息内容不是定长的。

单个文件大小默认1G, 文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。

每个Rocket实例只会往一个commitlog文件中写，写完一个接着写下一个。indexFile和ComsumerQueue中都有消息对应的物理偏移量，通过物理偏移量就可以计算出该消息位于哪个CommitLog文件上。

CommitLog目前存储的MappedFile文件有两种内容类型：

• MESSAGE：消息信息
• BLANK：文件不足以存储消息时的空白占位

MESSAGE

BLANK

ConsumeQueue

ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每个Topic下的每个Message Queue都有一个对应的 ConsumeQueue文件。

为了加速ConsumeQueue消息条目的检索速度与节省磁盘空间，每一个 Consumequeue条目不会存储消息的全量信息，存储结构如下：

ConsumeQueue是消息的逻辑队列，相当于字典的目录，用来指定消息在物理文件CommitLog上的位置。

ConsumeQueue对应一个topic的一个队列结构，由topic和queueId可以唯一创建一个ConsumeQueue结构，同样ConsumeQueue包含一个MappedFileQueue结构，而MappedFileQueue结构由多个MappedFile文件组成，每个文件的大小为30000020（300000 ConsumeQueue.CQ_STORE_UNIT_SIZE），其中20是queue中每个存储单元的大小。

consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：${user.home}/store/consumequeue/{topic}/{queueId}/{fileName}。

ConsumeQueue即为CommitLog文件的索引文件， 其构建机制是当消息到达Commitlog文件后 由专门的线程产生消息转发任务，从而构建消息消费队列文件（ConsumeQueue ）与索引文件（Index File）。

存储机制这样设计有以下几个好处：

1. CommitLog顺序写 ，可以大大提高写入效率。（实际上，磁盘有时候会比你想象的快很多，有时候也比你想象的慢很多，关键在如何使用，使用得当，磁盘的速度完全可以匹配上网络的数据传输速度。目前的高性能磁盘，顺序写速度可以达到600MB/s，超过了一般网卡的传输速度，这是磁盘比想象的快的地方，但是磁盘随机写的速度只有大概lOOKB/s，和顺序写的性能相差6000倍！）
2. 虽然是随机读，但是利用操作系统的pagecache机制，可以批量地从磁盘读取，作为cache存到内存中，加速后续的读取速度。
3. 为了保证完全的顺序写，需要ConsumeQueue这个中间结构，因为ConsumeQueue里只存偏移量信息，所以尺寸是有限的，在实际情况中，大部分的ConsumeQueue能够被全部读入内存，所以这个中间结构的操作速度很快，可以认为是内存读取的速度。此外为了保证CommitLog和ConsumeQueue的一致性，CommitLog里存储了Consume Queues、Message Key、Tag等所有信息，即使ConsumeQueue丢失，也可以通过CommitLog完全恢复出来。

IndexFile

RocketMQ还支持通过MessageID或者MessageKey来查询消息；使用ID查询时，因为ID就是用broker+offset生成的（这里msgId指的是服务端的），所以很容易就找到对应的commitLog文件来读取消息。但是对于用MessageKey来查询消息，RocketMQ则通过构建一个index来提高读取速度。

index存的是索引文件，这个文件用来加快消息查询的速度。消息消费队列RocketMQ专门为消息订阅构建的索引文件，提高根据主题与消息检索消息的速度 ，使用Hash索引机制，具体是Hash槽与Hash冲突的链表结构。

IndexFile 也是定长的，从单个文件的数据结构来说，这是实现了一种简单原生的哈希拉链机制。当一条新的消息索引进来时，首先使用 hash 算法命中黄色部分 500w 个 slot 中的一个，如果存在冲突就使用拉链解决，将最新索引数据的 next 指向上一条索引位置。同时将消息的索引数据 append 至文件尾部（绿色部分），这样便形成了一条当前 slot 按照时间存入的倒序的链表。

Index Header

Slot

Content

Config

config文件夹中存储着Topic和Consumer等相关信息。主题和消费者群组相关的信息就存在在此。

• topics.json : topic 配置属性
• subscriptionGroup.json :消息消费组配置信息。
• delayOffset.json ：延时消息队列拉取进度。
• consumerOffset.json ：集群消费模式消息消进度。
• consumerFilter.json ：主题消息过滤信息。

过期文件删除

由于RocketMQ操作CommitLog，ConsumeQueue文件是基于内存映射机制并在启动的时候会加载CommitLog，ConsumeQueue目录下的所有文件，为了避免内存与磁盘的浪费，不可能将消息永久存储在消息服务器上，所以需要引入一种机制来删除己过期的文件。

删除过程分别执行清理消息存储文件（Commitlog）与消息消费队列文件（ConsumeQueue文件），消息消费队列文件与消息存储文件共用一套过期文件机制。

RocketMQ清除过期文件的方法是：如果非当前写文件在一定时间间隔内没有再次被更新，则认为是过期文件，可以被删除，RocketMQ不会关注这个文件上的消息是否全部被消费。默认每个文件的过期时间为72小时（不同版本的默认值不同，这里以4.9.4为例），通过在Broker配置文件中设置 fileReservedTime来改变过期时间，单位为小时。

触发文件清除操作的是一个定时任务，而且只有定时任务，文件过期删除定时任务的周期由该删除决定，默认每10s执行一次。

过期判断

文件删除主要是由这个配置属性：fileReservedTime，文件保留时间。也就是从最后一次更新时间到现在，如果超过了该时间，则认为是过期文件，可以删除。

另外还有其他两个配置参数：

• deletePhysicFilesInterval：删除物理文件的时间间隔（默认是100MS），在一次定时任务触发时，可能会有多个物理文件超过过期时间可被删除，因此删除一个文件后需要间隔deletePhysicFilesInterval这个时间再删除另外一个文件，由于删除文件是一个非常耗费IO的操作，会引起消息插入消费的延迟（相比于正常情况下），所以不建议直接删除所有过期文件。
• destroyMapedFileIntervalForcibly：在删除文件时，如果该文件还被线程引用，此时会阻止此次删除操作，同时将该文件标记不可用并且纪录当前时间戳destroyMapedFileIntervalForcibly这个表示文件在第一次删除拒绝后，文件保存的最大时间，在此时间内一直会被拒绝删除，当超过这个时间时，会将引用每次减少1000，直到引用小于等于0为止，即可删除该文件。

删除条件

1. 指定删除文件的时间点，RocketMQ通过deleteWhen设置一天的固定时间执行一次。删除过期文件操作，默认为凌晨4点。
2. 磁盘空间是否充足，如果磁盘空间不充足(DiskSpaceCleanForciblyRatio。磁盘空间强制删除文件水位。默认是85)，会触发过期文件删除操作。

另外还有RocketMQ的磁盘配置参数：

1. 物理使用率大于diskSpaceWarningLevelRatio（默认90%可通过参数设置），则会阻止新消息的插入。
2. 物理磁盘使用率小于diskMaxUsedSpaceRatio(默认75%) 表示磁盘使用正常。

目录

一、概述

二、CommitLog文件

三、ConsumerQueue消费逻辑队列

四、IndexFile索引文件

五、页缓存与内存映射

一、概述

RocketMQ中的消息存储在本地文件系统中，主要是由ConsumeQueue和CommitLog配合完成的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。

来看一张RocketMQ消息存储整体架构图：

我们可以在broker的配置文件中配置store存储的目录。例如前面搭建RocketMQ时我们就自定义了存储目录。

#存储路径
storePathRootDir=/rocketmq/rocketmq-4.9.2/store
#commitLog 存储路径
storePathCommitLog=/rocketmq/rocketmq-4.9.2/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/rocketmq/rocketmq-4.9.2/store/consumequeue
#消息索引存储路径
storePathIndex=/rocketmq/rocketmq-4.9.2/store/index
#checkpoint 文件存储路径
storeCheckpoint=/rocketmq/rocketmq-4.9.2/store/checkpoint
#abort 文件存储路径
abortFile=/rocketmq/rocketmq-4.9.2/store/abort

分别介绍一下各个文件夹/文件的含义：

• abort：该文件在Broker启动后会自动创建，正常关闭Broker，该文件会自动消失。若在没有启动 Broker的情况下，发现这个文件是存在的，则说明之前Broker的关闭是非正常关闭；
• checkpoint：其中存储着commitlog、consumequeue、index文件的最后刷盘时间戳；
• commitlog：其中存放着commitlog文件，而消息是写在commitlog文件中的；
• config：存放着Broker运行期间的一些配置数据；
• consumequeue：其中存放着consumequeue文件，队列就存放在这个目录中；
• index：其中存放着消息索引文件indexFile；
  lock：运行期间使用到的全局资源锁；

二、CommitLog文件

CommitLog是消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容，消息内容不是定长的。单个文件大小默认1G, 文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件。

需要注意的是，一个Broker中仅包含一个commitlog目录，所有的mappedFile文件都是存放在该目录中 的。即无论当前Broker中存放着多少Topic的消息，这些消息都是被顺序写入到了mappedFile文件中。

RocketMQ利用“零拷贝”技术，提高消息存盘和网络发送的速度。这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因。

三、ConsumerQueue消费逻辑队列

ConsumerQueue是消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构：

如下图，可以看到一个consumequeue中的索引条目结构为：

同样consumequeue文件采取定长设计，单个consumequeue文件由30W个索引条目组成。每一个索引条目包含了三个重要属性：消息在commitlog中的偏移量（8字节）、消息长度（4字节）、消息Tag的hashcode值（8字节）。可以像数组一样随机访问每一个索引条目，每个索引条目这三个属性占20个字节，所以每个ConsumeQueue文件的大小是固定的30w * 20字节，约5.72M。

四、IndexFile索引文件

除了通过通常的指定Topic进行消息消费外，RocketMQ还提供了根据key进行消息查询的功能。该查询是通过store目录中的index子目录中的indexFile进行索引实现的快速查询。当然，这个indexFile中的索引数据是在包含了key的消息被发送到Broker时写入的。如果消息中没有包含key，则不会写入。

RocketMQ的索引文件逻辑结构，类似JDK中HashMap的实现。索引文件的具体结构如下：

每个Broker中会包含一组indexFile，每个indexFile都是以一个时间戳命名的（这个indexFile被创建时的时间戳），文件大小是固定的。如果消息的properties中设置了UNIQ_KEY这个属性，就用 topic + “#” + UNIQ_KEY的value作为 key 来做写入操作。如果消息设置了KEYS属性（多个KEY以空格分隔），也会用 topic + “#” + KEY 来做索引。

其中的索引数据包含了Key Hash、CommitLog Offset、Timestamp、NextIndex offset 这四个字段，一共20 字节。NextIndex offset 即前面读出来的slotValue，如果有 hash冲突，就可以用这个字段将所有冲突的索引用链表的方式串起来了。Timestamp记录的是消息storeTimestamp之间的差，并不是一个绝对的时间。整个Index File的结构如图，40 Byte 的Header用于保存一些总的统计信息，4*500W的 Slot Table并不保存真正的索引数据，而是保存每个槽位对应的单向链表的头。20*2000W 是真正的索引数据，即一个 Index File 可以保存 2000W个索引。

五、页缓存与内存映射

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

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

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