【RocketMQ】原理分析：消息存储机制

淡淡的烟草味﹌ 2022-12-22 08:50 131阅读 0赞

由于分布式消息队列对于可靠性的要求比较高，所以需要保证生产者将消息发送到broker之后，保证消息是不出现丢失的，因此消息队列就少不了对于可靠性存储的要求。

从主流的几种MQ消息队列采用的存储方式来看，主要会有三种

1.  分布式KV存储，
    
     *  这种存储方式对于消息读写能力要求不高的情况可以使用
     *  比如ActiveMQ中采用的levelDB、Redis，
2.  文件系统存储，
    
     *  这种方案适合对于有高吞吐量要求的消息中间件，因为**消息刷盘是**一种高效率，高可靠、高性能的持久化方式，除非磁盘出现故障，否则一般是不会出现无法持久化的问题
     *  常见的比如kafka、RocketMQ、RabbitMQ都是采用消息刷盘到所部署的机器上的文件系统来做持久化
3.  关系型数据库，
    
     *  关系型数据库在单表数据量达到千万级的情况下IO性能会出现瓶颈，
     *  比如ActiveMQ可以采用mysql作为消息存储，所以ActiveMQ并不适合于高吞吐量的消息队列场景。

总的来说，对于存储效率，文件系统要优于分布式KV存储，分布式KV存储要优于关系型数据库

## 1.消息存储的整体结构 ##

RocketMQ的消息存储采用的是混合型的存储结构，也就是Broker单个实例下的所有队列公用一个日志数据文件CommitLog。这个是和Kafka又一个不同之处。

> 为什么不采用kafka的设计，针对不同的partition存储一个独立的物理文件呢？  
> 这是因为在kafka的设计中，一旦kafka中Topic的Partition数量过多，队列文件会过多，那么会给磁盘的IO读写造成比较大的压力，也就造成了性能瓶颈。所以RocketMQ进行了优化，消息主题统一存储在CommitLog中。
> 
>  *  优点：
>     
>      *  由于消息主题都是通过CommitLog来进行读写，ConsumerQueue中只存储很少的数据， 所以队列更加轻量化。
>      *  对于磁盘的访问是串行化从而避免了磁盘的竞争
>  *  缺点：
>     
>      *  消息写入磁盘虽然是基于**顺序写**，但是读的过程确是**随机读**。
>      *  读取一条消息会先读取 ConsumeQueue，再读CommitLog，会降低消息读的效率。

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

## 2.消息的存储文件结构 ##

RocketMQ就是采用文件系统的方式来存储消息，消息的存储是由ConsumeQueue和CommitLog配合完成的。

*  CommitLog是消息真正的物理存储文件。
 *  ConsumeQueue是消息的逻辑队列，有点类似于数据库的索引文件，里面存储的是指向CommitLog文件中消息存储的地址。

RocketMQ的存储文件默认在`root/store`目录下，可以看到这样一个结构的文件。

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

我们只需要关心Commitlog、Consumequeue、Index

**1.CommitLog**

CommitLog是用来存放消息的物理文件，每个broker上的commitLog本当前机器上的所有 consumerQueue共享，不做任何的区分。即所有topic的数据都存在一起

*  关于文件大小
    
     *  CommitLog中的文件默认大小为1G，可以动态配置；
     *  当一个文件写满以后，会生成一个新的commitlog文件。所有的Topic数据是顺序写入在CommitLog文件中的。
 *  关于文件名
    
     *  文件名的长度为20位，左边补0，剩余未起始偏移量
     *  比如 00000000000000000000 表示第一个文件，
        
        当第一个文件写满后，生成第二个文件 000000000001073741824 ，起始偏移量为1073741824

**2.ConsumeQueue**

consumeQueue表示消息消费的逻辑队列，这里面包含MessageQueue在commitlog中的其实物理位置偏移量oﬀset，消息实体内容的大小和Message Tag的hash值。

*  关于文件大小
    
    对于实际物理存储来说， consumeQueue对应每个topic和queueid下的文件，每个consumeQueue类型的文件也是有大小，每个文件默认大小约为600W个字节，如果文件满了后会也会生成一个新的文件
 *  关于文件名
    
    每个Topic下的每个Message Queue都会对应一个ConsumeQueue文件，比如上图中的testCreateTopic3主题下有2个文件夹，分别代表消息队列0和1。
    
    每个消息队列的文件地址是：`root/store/consumequeue/{topicNmae}/{queueId}/{ﬁlename}`

**3.IndexFile**

索引文件，如果一个消息包含Key值的话，会使用IndexFile存储消息索引。Index索引文件提供了对 CommitLog进行数据检索，提供了一种通过key或者时间区间来查找CommitLog中的消息的方法。

在物理存储中，文件名是以创建的时间戳明明，固定的单个IndexFile大小大概为400M，一个IndexFile可以保存2000W个索引

**4.abort**

broker在启动的时候会创建一个空的名为abort的文件，并在shutdown时将其删除，用于标识进程是否正常退出，如果不正常退出,会在启动时做故障恢复

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/c7d3f5185eeb4d19a4d9bf0685e6b0dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/79c5d9d4d50d4ac0a87d63122c2f1175.png