正文 

当前位置： 首页 > 代码 > java教程

一文详解rocketmq的存储模型

author author     2023-01-06     206

关键词：

摘要：RocketMQ 优异的性能表现，必然绕不开其优秀的存储模型。

本文分享自华为云社区《​​终于弄明白了 RocketMQ 的存储模型​​》，作者：勇哥java实战分享。

RocketMQ 优异的性能表现，必然绕不开其优秀的存储模型 。

1 整体概览

首先温习下 RocketMQ 架构。

一文详解RocketMQ的存储模型_消息队列

整体架构中包含四种角色 :

  • Producer ：消息发布的角色，Producer 通过 MQ 的负载均衡模块选择相应的 Broker 集群队列进行消息投递，投递的过程支持快速失败并且低延迟。
  • Consumer ：消息消费的角色，支持以 push 推，pull 拉两种模式对消息进行消费。
  • NameServer ：名字服务是一个非常简单的 Topic 路由注册中心，其角色类似 Dubbo 中的 zookeeper ，支持 Broker 的动态注册与发现。
  • BrokerServer ：Broker 主要负责消息的存储、投递和查询以及服务高可用保证 。

本文的重点在于分析 BrokerServer 的消息存储模型。我们先进入 broker 的文件存储目录 。

一文详解RocketMQ的存储模型_偏移量_02

消息存储和下面三个文件关系非常紧密：

1.数据文件 commitlog

消息主体以及元数据的存储主体 ；

2.消费文件 consumequeue

消息消费队列，引入的目的主要是提高消息消费的性能 ；

3.索引文件 index

索引文件，提供了一种可以通过 key 或时间区间来查询消息。

RocketMQ 采用的是混合型的存储结构，Broker 单个实例下所有的队列共用一个数据文件（commitlog）来存储。

生产者发送消息至 Broker 端，然后 Broker 端使用同步或者异步的方式对消息刷盘持久化，保存至 commitlog 文件中。只要消息被刷盘持久化至磁盘文件 commitlog 中，那么生产者发送的消息就不会丢失。

Broker 端的后台服务线程会不停地分发请求并异步构建 consumequeue（消费文件）和 indexFile（索引文件）。

2 数据文件

RocketMQ 的消息数据都会写入到数据文件中， 我们称之为 commitlog 。

所有的消息都会顺序写入数据文件，当文件写满了，会写入下一个文件。

一文详解RocketMQ的存储模型_数据文件_03

如上图所示，单个文件大小默认 1G , 文件名长度为 20 位，左边补零，剩余为起始偏移量，比如 00000000000000000000 代表了第一个文件，起始偏移量为 0 ，文件大小为1 G = 1073741824。

当第一个文件写满了，第二个文件为 00000000001073741824，起始偏移量为 1073741824，以此类推。

一文详解RocketMQ的存储模型_存储模型_04

从上图中，我们可以看到消息是一条一条写入到文件，每条消息的格式是固定的。

这样设计有三点优势：

1、顺序写

磁盘的存取速度相对内存来讲并不快，一次磁盘 IO 的耗时主要取决于：寻道时间和盘片旋转时间，提高磁盘 IO 性能最有效的方法就是：减少随机 IO，增加顺序 IO 。

一文详解RocketMQ的存储模型_RocketMQ_05

《 The Pathologies of Big Data 》这篇文章指出：内存随机读写的速度远远低于磁盘顺序读写的速度。磁盘顺序写入速度可以达到几百兆/s，而随机写入速度只有几百 KB /s，相差上千倍。

2、快速定位

因为消息是一条一条写入到 commitlog 文件 ，写入完成后，我们可以得到这条消息的物理偏移量。

每条消息的物理偏移量是唯一的， commitlog 文件名是递增的，可以根据消息的物理偏移量通过二分查找，定位消息位于那个文件中，并获取到消息实体数据。

3、通过消息 offsetMsgId 查询消息数据

一文详解RocketMQ的存储模型_存储模型_06

消息 offsetMsgId 是由 Broker 服务端在写入消息时生成的 ，该消息包含两个部分：

  • Broker 服务端 ip + port 8个字节；
  • commitlog 物理偏移量 8个字节 。

我们可以通过消息 offsetMsgId ，定位到 Broker 的 ip 地址 + 端口 ，传递物理偏移量参数 ，即可定位该消息实体数据。

3 消费文件

在介绍 consumequeue 文件之前， 我们先温习下消息队列的传输模型-发布订阅模型 ， 这也是 RocketMQ 当前的传输模型。

一文详解RocketMQ的存储模型_存储模型_07

发布订阅模型具有如下特点：

  • 消费独立：相比队列模型的匿名消费方式，发布订阅模型中消费方都会具备的身份，一般叫做订阅组（订阅关系），不同订阅组之间相互独立不会相互影响。
  • 一对多通信：基于独立身份的设计，同一个主题内的消息可以被多个订阅组处理，每个订阅组都可以拿到全量消息。因此发布订阅模型可以实现一对多通信。

因此，rocketmq 的文件设计必须满足发布订阅模型的需求。

那么仅仅 commitlog 文件是否可以满足需求吗 ？

假如有一个 consumerGroup 消费者，订阅主题 my-mac-topic ，因为 commitlog 包含所有的消息数据，查询该主题下的消息数据，需要遍历数据文件 commitlog , 这样的效率是极其低下的。

进入 rocketmq 存储目录，显示见下图：

一文详解RocketMQ的存储模型_数据文件_08

  1. 消费文件按照主题存储，每个主题下有不同的队列，图中 my-mac-topic 有 16 个队列 ;
  2. 每个队列目录下 ，存储 consumequeue 文件，每个 consumequeue 文件也是顺序写入，数据格式见下图。

一文详解RocketMQ的存储模型_RocketMQ_09

每个 consumequeue 包含 30 万个条目，每个条目大小是 20 个字节，每个文件的大小是 30 万 * 20 = 60万字节，每个文件大小约5.72M 。和 commitlog 文件类似，consumequeue 文件的名称也是以偏移量来命名的，可以通过消息的逻辑偏移量定位消息位于哪一个文件里。

消费文件按照主题-队列来保存 ，这种方式特别适配发布订阅模型。

消费者从 broker 获取订阅消息数据时，不用遍历整个 commitlog 文件，只需要根据逻辑偏移量从 consumequeue 文件查询消息偏移量 , 最后通过定位到 commitlog 文件， 获取真正的消息数据。

这样就可以简化消费查询逻辑，同时因为同一主题下，消费者可以订阅不同的队列或者 tag ，同时提高了系统的可扩展性。

4 索引文件

每个消息在业务层面的唯一标识码要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过 topic、key 来查询这条消息内容，以及消息被谁消费。

由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。

//订单Id   
String orderId = "1234567890"; 
message.setKeys(orderId);

从开源的控制台中根据主题和 key 查询消息列表：

一文详解RocketMQ的存储模型_偏移量_10

进入索引文件目录 ，如下图所以：

一文详解RocketMQ的存储模型_数据文件_11

索引文件名 fileName 是以创建时的时间戳命名的，固定的单个 IndexFile 文件大小约为 400 M 。

IndexFile 的文件逻辑结构类似于 JDK 的 HashMap 的数组加链表结构。

一文详解RocketMQ的存储模型_RocketMQ_12

索引文件主要由 Header、Slot Table (默认 500 万个条目)、Index Linked List（默认最多包含 2000万个条目）三部分组成 。

一文详解RocketMQ的存储模型_消息队列_13

假如订单系统发送两条消息 A 和 B , 他们的 key 都是 “1234567890” ，我们依次存储消息 A , 消息 B 。

因为这两个消息的 key 的 hash 值相同，它们对应的哈希槽（深黄色）也会相同，哈希槽会保存的最新的消息 B 的索引条目序号 , 序号值是 4 ，也就是第二个深绿色条目。

而消息 B 的索引条目信息的最后 4 个字节会保存上一条消息对应的索引条目序号，索引序号值是 3 , 也就是消息 A 。

5 写到最后

Databases are specializing – the “one size fits all” approach no longer applies ------ MongoDB设计哲学

RocketMQ 存储模型设计得非常精巧，笔者觉得每种设计都有其底层思考，这里总结了三点 ：

  1. 完美适配消息队列发布订阅模型 ；
  2. 数据文件，消费文件，索引文件各司其职 ，同时以数据文件为核心，异步构建消费文件 + 索引文件这种模式非常容易扩展到主从复制的架构；
  3. 充分考虑业务的查询场景，支持消息 key ，消息 offsetMsgId 查询消息数据。也支持消费者通过 tag 来订阅主题下的不同消息，提升了消费者的灵活性。


​​点击关注，第一时间了解华为云新鲜技术~​​

一文详解rocketmq的存储模型(代码片段)

摘要：RocketMQ优异的性能表现，必然绕不开其优秀的存储模型。本文分享自华为云社区《终于弄明白了RocketMQ的存储模型》，作者：勇哥java实战分享。RocketMQ优异的性能表现，必然绕不开其优秀的存储模型。1整... 查看详情

rocketmq源码学习--消息存储篇

1.序言今天来和大家探讨一下RocketMQ在消息存储方面所作出的努力，在介绍RocketMQ的存储模型之前，可以先探讨一下MQ的存储模型选择。2.MQ的存储模型选择个人看来，从MQ的类型来看，存储模型分两种：需要持久化（ActiveMQ,RabbitMQ,K... 查看详情

一文详解，rocketmq事务消息

在RocketMQ中有一个非常有用的功能，就是事务消息功能，事务消息机制，可以让我们确保发送的消息一定能写进MQ里，绝不会丢失掉。MQ事务消息机制还是挺有用的，在业内还是比较常见的，所以今天我们就... 查看详情

（转）rocketmq源码学习--消息存储篇

...tp://www.tuicool.com/articles/umQfMzA1.序言今天来和大家探讨一下RocketMQ在消息存储方面所作出的努力，在介绍RocketMQ的存储模型之前，可以先探讨一下MQ的存储模型选择。2.MQ的存储模型选择个人看来，从MQ的类型来看，存储模型分两种：... 查看详情

rocketmq消息队列——消息存储详解(代码片段)

2016年双11前后阿里巴巴将RocketMQ捐赠给Apache基金会，很快就吸引了全球众多开源爱好者加入其社区生态中，并在2017年9月成为Apache基金会的顶级项目。利用RocketMQ可以轻松实现应用解耦、流量消峰、消息分发等功能，并... 查看详情

rocketmq消息队列——消息存储详解(代码片段)

2016年双11前后阿里巴巴将RocketMQ捐赠给Apache基金会，很快就吸引了全球众多开源爱好者加入其社区生态中，并在2017年9月成为Apache基金会的顶级项目。利用RocketMQ可以轻松实现应用解耦、流量消峰、消息分发等功能，并... 查看详情

一文详解扩散模型：ddpm(代码片段)

人工智能生成内容（AIGeneratedContent，AIGC）近年来成为了非常前沿的一个研究方向，生成模型目前有四个流派，分别是生成对抗网络（GenerativeAdversarialModels，GAN），变分自编码器（VarianceAuto-Encoder，VAE），标准化流模型（Normalizatio... 查看详情

推荐收藏一文详解数据分析经典模型rfm(附python代码)(代码片段)

RFM（RecencyFrequencyMonetary）模型是衡量客户价值和客户创利能力的重要工具和手段。在众多的客户关系管理(CRM)的分析模式中，RFM模型是被广泛提到的，它的本质是用户分类。本文将用Python语言来讲解的RFM模型，... 查看详情

「软件项目管理」一文详解软件项目质量计划(代码片段)

一文详解软件项目质量计划🎩前言👒一、质量概述1.质量与软件质量2.质量成本二、质量模型1.定义2.几种模型3.模型解读（1）Bohem质量模型（2）McCall质量模型（3）ISO/IEC9126质量模型（4）ISO/I... 查看详情

rocketmq详解(代码片段)

...：http://www.cnblogs.com/xiaodf/p/5075167.html简介官方简介： RocketMQ是一款分布式、队列模型的消息中间件，具有以下特点： 能够保证严格的消息顺序 提供丰富的消息拉取模式 高效的订阅者水平扩展能力 实时的消息... 查看详情

rocketmq：消息存储机制详解与源码解析(代码片段)

文章目录消息存储机制1.前言2.核心存储类：DefaultMessageStore3.消息存储流程4.消息存储文件5.存储文件内存映射5.1.MapperFileQueue5.2.MappedFile5.2.1.commit5.2.2.flush5.3.TransientStorePool6.刷盘机制6.1.同步刷盘6.2.异步刷盘消息存储机制1.前言... 查看详情

rocketmq详解系列(代码片段)

什么是RocketMQRocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。主要功能是异步解耦和流量削峰：。常见的MQ主要有：ActiveMQ、RabbitMQ、Kafka、Ro... 查看详情

rocketmq详解系列(代码片段)

什么是RocketMQRocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。主要功能是异步解耦和流量削峰：。常见的MQ主要有：ActiveMQ、RabbitMQ、Kafka、Ro... 查看详情

rocketmq详解系列(代码片段)

什么是RocketMQRocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。主要功能是异步解耦和流量削峰：。常见的MQ主要有：ActiveMQ、RabbitMQ、Kafka、Ro... 查看详情

mysql深潜-一文详解mysqldatadictionary(代码片段)

简介： 在MySQL8.0之前，Server层和存储引擎（比如InnoDB）会各自保留一份元数据（schemaname,tabledefinition等），不仅在信息存储上有着重复冗余，而且可能存在两者之间存储的元数据不同步的现象。不... 查看详情

消息中间件-rocketmq详解（从软件安装到案例实现）(代码片段)

RocketMQ内容一、RocketMQ安装二、RocketMQ作用和结构1.RocketMQ特点2RocketMQ执行流程3.RocketMQ作用3.1消息中间件结构图3.2应用解耦3.3削峰填谷4.rocketmq组成部分5.rocketmq基本概念模型三、生产消息的类型有三种四、消费模式有两种五、延时... 查看详情

一文带你理解rocketmq广播模式实现机制

RocketMQ有两种消费模式，集群模式和广播模式。集群模式是指RocketMQ中的一条消息只能被同一个消费者组中的一个消费者消费。如下图，Producer向TopicTest这个Topic并发写入3条新消息，分别被分配到了MessageQueue1~MessageQueue3这3个队列... 查看详情

深度解读rocketmq存储机制

RocketMQ实现了灵活的多分区和多副本机制，有效的避免了集群内单点故障对于整体服务可用性的影响。存储机制和高可用策略是RocketMQ稳定性的核心，社区上关于RocketMQ目前存储实现的分析与讨论一直是一个热议的话题。... 查看详情