RocketMQ源码：broker存储文件组织和内存映射

拼搏现实的明天。 2024-03-25 22:31 92阅读 0赞

## 1. 存储设计 ##

`RocketMQ`存储文件主要包括：`CommitLog`文件、`ConsumerQueue`文件、`Index`文件

*  `CommitLog`文件：所有Topic的消息按照抵达`顺序`依次追加到`CommitLog`中，一旦写入不支持修改
 *  `ConsumeQueue`文件：消息消费队列，`用于消费者消费`，即消费者通过此文件来从`CommitLog`中获取消息。消息达到`CommitLog`后，将异步转发到`ConsumeQueue`文件
 *  `Index`文件：消息索引，主要存储`消息key与offset`的对应关系

RocketMQ将所有`Topic`的消息都存储在`同一个CommitLog`文件中，一般按照`Topic`来检索消息，所以为了提高消息消费的效率，`RocketMQ`引入了`ConsumeQueue`文件（消费队列），每一个`Topic`包含多个消息消费队列，每一个消费队列都有一个文件。

为了根据消息的属性从`CommitLog`文件中快速检索消息，RocketMQ引入了`Index`索引文件。

存储目录为：

![format_png][]

### 1.1 基本原理 ###

*  `Page Cache`:`Page Cache`是文件系统层的`Cache`，主要用来`减少对磁盘的I/O操作`，通过对磁盘的访问变为物理内存的访问，缓存的是内存页面，操作时按照页为基本单位。在Linux系统中写入数据的时候并不会直接写到硬盘上，而是会先写到`Page Cache`中，并打上`dirty`标识，由内核线程`flusher`定期将被打上`dirty`的页发送给IO调度层，最后由IO调度决定何时落地到磁盘中，而Linux一般会把还没有使用的内存全拿来给`Page Cache`使用。而读的过程也是类似，会先到`Page Cache`中寻找是否有数据，有的话直接返回，如果没有才会到磁盘中去读取并写入`Page Cache`，然后再次读取`Page Cache`并返回。而且读的这个过程中操作系统也会有一个预读的操作，你的每一次读取操作系统都会帮你预读出后面一部分数据。当你一直在使用预读数据的时候，系统会帮你预读出更多的数据(最大到128K)。
 *  `Buffer Cache`:`Buffer Cache`是针对设备的，实际操作按块为基本单位，对于裸盘的读写会占用`Buffer Cache`，当读写完成之后，会归还给操作系统。
 *  在linux2.4内核中`Buffer Cache`和`Page Cache`是共存的，因为文件的读写最终会转化为块设备的读写，即同一份文件的数据，可能既在`Buffer Cache`中也在`Page Cache`中，这样就造成了物理内存的浪费。
 *  linux2.6内核对`Buffer Cache`和`Page Cache`进行了合并，统一为`Page Cache`。当进行文件读写时，如果文件在磁盘上的存储块是连续的，那么文件在`Page Cache`中对应的页是普通的page，如果文件在磁盘上的数据块是不连续的，或者是设备文件，那么文件在`Page Cache`中对应的页就是`Buffer Cache`
 *  查看内存情况
    
        $ # cat /proc/meminfo                     
        MemTotal:        3876772 kB
        MemFree:          126704 kB
        MemAvailable:     137132 kB
        Buffers:              48 kB
        Cached:           258648 kB
        SwapCached:        12344 kB
        ...省略...
        
        Buffers: 表示`Buffer Cache`的容量
        Cached: 表示位于物理内存中的页缓存`Page Cache`
        SwapCached:表示位于磁盘交换区的页缓存`Page Cache`
        实际的`Page Cache`容量=Cached+SwapCached
        复制代码
 *  linux底层提供`mmap将文件映射进虚拟内存`，对文件的读写变成对内存的读写，能充分利用`Page Cache`,但是如果对文件进行随机读写，会使虚拟内存产生很多缺页(`Page Fault`)中断，此时操作系统需要将磁盘文件的数据再次加载到`Page Cache`，这个过程比较慢。如果对文件进行顺序读写，读和写的区域都是被操作系统缓存过的热点区域，不会产生大量的缺页中断，文件的读写操作相当于直接内存的操作，性能会提升很多。如果内存不够充足，内核把内存分配给`Page Cache`后，空闲内存会变少，如果程序有新的内存分配或者缺页中断，但是空闲内存不够，内核需要花费时间将热度低的`Page Cache`内存回收掉，此时性能会下降。当遇到操作系统进行脏页回写，内存回收，内存换入换出等情形时，会产生较大的读写延迟，造成存储引擎偶发的高延迟，针对这种现象，RocketMQ采用了多种优化技术，比如内存预分配，文件预热，mlock系统调用，读写分离等，来保证利用`Page Cache`优点的同时，消除其带来的延迟。
 *  工具查看：`hcache`是基于`pcstat，pcstat`可以查看文件是否被缓存和根据pid来查看缓存了哪些文件，`hcache`是`pcstat`的增强版本，增加了查看整个系统`Cache`和根据`Cache`大小排序的功能：
    
        查看使用Cache最多的3个进程
        $ hcache --top 3  
        +----------------------------------+--------------+-------+--------+---------+
        | Name                             | Size (bytes) | Pages | Cached | Percent |
        |----------------------------------+--------------+-------+--------+---------|
        | /usr/share/atom/atom             | 81137776     | 19810 | 19785  | 099.874 |
        | /usr/bin/dockerd                 | 68608880     | 16751 | 14321  | 085.493 |
        | /usr/share/atom/snapshot_blob.bin| 54619240     | 13335 | 13335  | 100.000 |
        +----------------------------------+--------------+-------+--------+---------+
        复制代码

### 1.2 CommitLog ###

*  `RocketMQ Broker`单个实例下所有的`Topic`都使用同一个日志数据文件(`CommitLog`)来存储(即单个实例消息整体有序)，这点与kafka不同(kafka采用每个分区一个日志文件存储)
 *  `CommitLog`单个文件大小`默认1G`，文件文件名是`起始偏移量`，总共20位，左边补零，起始偏移量是0。假设文件按照默认大小1G来算
    
     *  第一个文件的文件名为00000000000000000000 ，当第一个文件被写满之后，开始写入第二个文件
     *  第二个文件的文件名为00000000001073741824 ，`1G=1073741824=1024*1024*1024`
     *  第三个文件的文件名是00000000002147483648，(文件名相差`1G=1073741824=1024*1024*1024`)
 *  `CommitLog`按照上述命名的好处是给出任意一个消息的物理偏移量，可以`通过二分法进行查找`，快速定位这个文件的位置，然后用消息物理偏移量减去所在文件的名称，得到的差值就是在该文件中的绝对地址

### 1.3 ConsumeQueue ###

`ConsumeQueue`是消息消费队列，它是一个逻辑队列，相当于`CommitLog的索引文件`。因为RocketMQ的队列不存储任何实际数据，它只存储`CommitLog`中的`【起始物理位置偏移量，消息的内容大小，消息Tag的哈希值】`，每一个`ConsumeQueue`存储的格式如下，`总共20B`。存tag是为了在消费者取到消息offset后先根据tag做一次过滤，剩下的才需要到`CommitLog`中取消息详情:

![format_png 1][]

每个`ConsumeQueue`都有一个`queueId`，queueId 的值为0到`TopicConfig`配置的队列数量。比如某个`Topic`的消费队列数量为4，那么四个`ConsumeQueue`的`queueId`就分别为0、1、2、3。

消费者消费时会先从`ConsumeQueue`中查找消息在`CommitLog`中的`offset`，再去`CommitLog`中找原始消息数据。如果某个消息只在`CommitLog`中有数据，没在`ConsumeQueue`中， 则消费者无法消费

`ConsumeQueue`类对应的是每个`topic和queuId`下面的所有文件。默认存储路径是`$HOME/store/consumequeue/{topic}/{queueId}/{fileName}`，每个文件由30w条数据组成，单个文件的大小是`30w x 20Byte`，即每个文件为600w字节，单个消费队列的文件大小约为`5.722M=(600w/(1024*1024))`

### 1.4 Index文件 ###

`IndexFile`:索引文件，物理存储上，`文件名为创建时间的时间戳命名`，固定的单个`IndexFile`文件大小约为400M，一个`IndexFile`可以保存`2000W个索引`

`IndexFile`（索引文件）由`IndexHeader`(索引文件头), `Slot`(槽位)和`Index`(消息的索引内容)三部分构成。对于每个`IndexFile`来说`IndexHeader`是固定大小的，`Slot`是索引的目录，用于定位`Index`在`IndexFile`中存储的物理位置。存储图：

![format_png 2][]

### 1.5 checkpoint文件 ###

`checkpoint`检查点文件的作用是记录`CommitLog、ConsumeQueue、Index文件的刷盘时间点`，文件固定长度为4kb，只用该文件的前24个字节

*  `physicMsgTimestamp`：CommitLog文件刷盘时间点
 *  `logicsMsgTimestamp`：ConsumeQueue文件刷盘时间点
 *  `indexMsgTimestamp`：Index文件刷盘时间点

### 1.6 TransientStorePool机制 ###

RocketMQ为了降低`PageCache`的使用压力，引入了`transientStorePoolEnable`机制，即`内存级别的读写分离机制`

默认情况，RocketMQ将消息写入`PageCache`，消费时从`PageCache`中读取消息。但是这样在高并发下`PageCache`压力会比较大，容易出现瞬时`broker busy`异常。RocketMQ通过开启`transientStorePoolEnable=true`，将`消息写入堆外内存`并立即返回，然后`异步将堆外内存中的数据批量提交到PageCache`，再`异步刷盘到磁盘`中。这样的好处就是形成内存级别的读写分离，发送写入消息是向堆外内存，消费读取消息是从PageCache

该机制的`缺点就是如果意外导致broker进程异常退`出，已经放入到PageCache中的数据不会丢失，而`存储在堆外内存的数据会丢失`

## 2. MappedFileQueue ##

RocketMQ使用`MappedFile、MappedFileQueue`来封装存储文件。`MappedFileQueue`是`MappedFile`的管理容器，使用`CopyOnWriteArrayList`来管理所有的`MappedFile`。`MappedFileQueue`提供查找目录下`Map`

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/68b7831a9a3346e7916755b5175cd1a5.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/3936987a22e44116bacff7e65808c521.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/e93f445c85f84af7b93435e9987fb8d4.png