乐趣区

关于rocketmq:一张图进阶-RocketMQ-消息存储

jiezi

前言

三此君看了好几本书，看了很多遍源码整顿的 一张图进阶 RocketMQ 图片，对于 RocketMQ 你只须要记住这张图！感觉不错的话，记得点赞关注哦。

【重要】视频在 B 站同步更新，欢送围观，轻轻松松涨姿态。一张图进阶 RocketMQ- 音讯存储（视频版）

https://www.bilibili.com/vide…

本文是“一张图进阶 RocketMQ”第 5 篇，对 RocketMQ 不理解的同学能够先看看后面 4 期：

  1. 一张图进阶 RocketMQ- 整体架构
  2. 一张图进阶 RocketMQ – NameServer
  3. 一张图进阶 RocketMQ – 音讯发送
  4. 一张图进阶 RocketMQ – 通信机制

后面两期咱们次要分享了 RocketMQ 是如何将音讯发送进来的，当初音讯曾经被 Netty 送上路了，接力棒曾经交给了 Broker。如果咱们本人来实现 Broker 会怎么实现呢？首先必定得把音讯存起来吧，不然宕机了，音讯失落了，那就离大谱了。

可是音讯要以什么构造存储呢？二进制、JSON、PB？从性能上来看必定都是能够的，那 RocketMQ 到底是怎么搞的？

解决了存储构造问题，那音讯存到哪里呢？数据库，本地文件，还是对象存储服务器？从性能的角度必定也都是能够的。可是，哪家数据库能够反对单机十万级吞吐量？那我间接通通存到数据库得了，瞎折腾些啥。难道存在本地文件就能够了？咱们本人实现不能够，然而 RocketMQ 能够，那 RocketMQ 有什么黑科技呢？

所以咱们明天就来聊一聊 Broker 如何存储音讯，【首先明确咱们的指标】咱们须要先理解 RocketMQ 的存储构造，也就是音讯是如何组织的。理解了存储构造，咱们能力更好的了解存储流程，不然咱们不晓得为什么流程是这样的。最初咱们须要理解有哪些机制撑持 RocketMQ 单机十万级吞吐量。

存储架构

音讯在 Broker 上的存储构造如上图，所有相干文件放在 ROCKETMQ_HOME 下，有哪些文件呢？寄存音讯自身的 CommitLog，以及音讯的索引文件 ConsumeQueue 和 IndexFile：

  • CommitLog

从物理构造上来看，所有的音讯都存储在 CommitLog 外面，其实就是所有的音讯依照“音讯在 CommitLog 各字段示意图”所示，挨个按顺序存储到文件中。

单个 CommitLog 文件大小默认 1G，文件名长度为 20 位，右边补零，残余为起始偏移量。比方 00000000000000000000 代表了第一个文件，起始偏移量为 0，文件大小为 1G=1073741824；当第一个文件写满了，第二个文件为 00000000001073741824，起始偏移量为 1073741824，以此类推。音讯次要是程序写入日志文件，当文件满了，写入下一个文件。CommitLog 程序写，能够大大提高写入效率。

然而问题来了，音讯发送的时候咱们指定了 Topic，当初所有 Topic 都程序个写入到 CommitLog，存入的时候是劳碌了（程序写），然而获取音讯可就麻烦了。如果我要获取某个 Topic 的音讯，须要遍历 commitlog 文件，依据 topic 过滤音讯。CommitLog 这个渣男，只管本人爽。有什么方法能够进步音讯查问效率呢？

  • ConsumeQueue

咱们再回顾一下，音讯存入的时候是指定了 Topic，同时咱们也说了每个 Topic 会对应多个 ConsumeQueue（queueId 标识）。要害就在 ConsumeQueue 上，ConsumeQueue 是指定 Topic 音讯的索引文件，怎么了解呢？从“音讯在 ConsumeQueue 各字段示意图”可知，每个条目共 20 个字节，别离为 8 字节的 commitlog 物理偏移量、4 字节的音讯长度、8 字节 tag hashcode，单个文件由 30W 个条目组成，能够像数组一样随机拜访每一个条目，每个 ConsumeQueue 文件大小约 5.72M。ConsumeQueue 文件能够看成是基于 topic 的 commitlog 索引文件。Consumer 即可依据 ConsumeQueue 来查找待生产的音讯。

因为 ConsumeQueue 里只存偏移量信息，所以尺寸是无限的，在理论状况中，大部分的 ConsumeQueue 可能被全副读入内存，所以这个两头构造的操作速度很快，能够认为是内存读取的速度。此外为了保障 CommitLog 和 ConsumeQueue 的一致性，CommitLog 里存储了 ConsumeQueues、Message Key、Tag 等所有信息，即便 ConsumeQueue 失落，也能够通过 CommitLog 完全恢复进去。
ConsumeQueue 文件夹的组织形式如下：topic/queue/file 三层组织构造，具体存储门路为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。

  • IndexFile

IndexFile 是另一种可选索引文件，提供了一种能够通过 key 或工夫区间来查问音讯的办法。IndexFile 索引文件其底层实现为 hash 索引，相似于 Java 1.7 HashMap，计算 Key 的 hashcode，hashcode 取余失去 hash 槽，拉链法解决哈希抵触。Index 文件的存储地位是：$HOME \store\index${fileName}，文件名 fileName 是以创立时的工夫戳命名的，固定的单个 IndexFile 文件大小约为 400M，一个 IndexFile 能够保留 2000W 个索引。

所以，RocketMQ 音讯存储架构次要有 CommitLog，ConsumeQueue，IndexFile 形成。咱们发送一条音讯，会先格式化成“音讯在 CommitLog 各字段示意图”中的样子，程序写入 CommitLog 中，而后 Broker 会依照”音讯在 ConsumeQueue 各字段示意图“所示构建一条索引记录，存入该音讯所属 Topic 的 ConsumeQueue 索引文件中。如果有 IndexFile，还会构建 IndexFile。

当初咱们曾经晓得了 RocketMQ 音讯的存储构造，接下来咱们的就要理解 RocketMQ 是如何构建 CommitLog、ConsumeQueue 和 IndexFile，以及 RocketMQ 如何保障性能，撑持单机十万级吞吐量的？这是本文的次要指标，肯定要抓住次要指标，不要走丢咯。

启动流程

理解了 RocketMQ 音讯在磁盘中是怎么存储的，咱们就能够来看看具体的存储流程了。首先，还是先来看看 Broker 的启动流程。初始化过程都是这个鸟样，只看初始化过程齐全不知所云，然而不看初始化过程，间接看具体执行流程也是摸不着头脑，一堆组件不晓得从哪里来的，所以咱们还是先耐着性子大抵看看。但这并不是咱们关注的重点，留神几个关键点即可。

  • 初始化启动环境。部署好 RocketMQ 后，执行 /bin/mqbroker 脚本，次要用于设置 RocketMQ 目录环境变量，例如 ROCKETMQ_HOME。而后调用 ./bin/runbroker.sh 进入 RocketMQ 的启动入口，次要设置了 JVM 启动参数，比方 JAVA_HOME、Xms、Xmx。执行 main 函数。
  • 初始化 BrokerController。该初始化次要蕴含 RocketMQ 启动命令行参数解析、NettyRemotingServer 初始化、Broker 各个模块配置参数解析、Broker 各个模块初始化、过程关机 Hook 初始化等过程。

  • 启动 RocketMQ 的各个组件。然而这些组件并不是每一个都是外围组件，局部组件会在前面的流程中应用，这里混个眼生，如果前面流程没有提及的大家能够暂且跳过，咱们的指标是把握 RocketMQ 的核心内容，而不是每个细节。

    • MessageStore：存储层服务，比方 CommitLog、ConsumeQueue 存储管理，音讯刷盘，索引构建 等。
    • RemotingServer：一般通道申请解决服务。个别的申请都是在这里被解决的。
    • FastRemotingServer：VIP 通道申请解决服务。如果一般通道比较忙，那么能够应用 VIP 通道，个别作为客户端降级应用。
    • BrokerOuterAPI：Broker 拜访对外接口的封装对象。
    • PullRequestHoldService：Pull 长轮询服务。
    • ClientHousekeepingService：清理心跳超时的生产者、消费者、过滤服务器。
    • FilterServerManager：过滤服务器治理。

    存储流程

    在后面 RocketMQ 存储构造中咱们理解了 RocketMQ 将所有音讯程序写入 CommitLog，而后构建 ConsumeQueue/IndexFile 索引文件，所以这个小结咱们次要的指标就是看看这些文件是如何构建的。

  • Broker 启动流程中很要害的一点是启动了 NettyRemotingServer，在 RocketMQ 通信机制（视频）中咱们介绍过 Broker(NettyRemotingServer) 初始化会监听端口期待客户端连贯，当客户端发送申请的时，NettyRemotingServer WorkerGroup 解决可读事件，调用 NettyServerHandler.channelRead0() 解决数据。
    接着调用链到 processRequestCommand 办法，这个办法次要是依据申请中的 RequestCode，从本地缓存 processorTable 中获取相应的 Processor 来执行后续逻辑。处理器是什么？处理器的缓存从哪里来？

    Processor 就是用来解决特定申请的执行者，例如，生产者存入音讯应用 SendMessageProcessor，查问音讯应用 QueryMessageProcessor，拉取音讯应用 PullMessageProcessor。在 Broker 启动流程中有一步是注册 Processor，以 RequestCode 为 Key，Processor 为值，增加到 processorTable 缓存中。接着 RocketMQ 音讯发送（视频）流程来看，当生产者的申请达到 Broker，Broker 获取的 Processor 应为 SendMessageProcessor。封装一个 Runable 对象，run 办法内调用 SendMessageProcessor.processRequest，提交到线程池，持续前面的解决。

  • SendMessageProcessor.processRequest 调用 sendMessage 办法，次要蕴含音讯的校验及重试逻辑解决，而后调用存储模块 DefaultMessageStore 存储音讯。
    音讯校验：校验 Broker 是否配置可写，校验 Topic 名字是否为默认值，获取或创立 topicConfig，判断 queueId 是否超过限度。
    重试音讯解决：消费者生产失败后会将音讯发回给 Broker，这里咱们暂且认为就是生产者发送的申请，先看上面的流程。
  • DefaultMessageStore.putMessage 只是做了很多的校验，简略看看即可。包含：如果以后 Broker 进行工作则回绝音讯写入、Broker 为 SLAVE 角色则回绝音讯写入、以后 RocketMQ 不反对写入则回绝音讯写入、主题长度超过 256 个字符则回绝音讯写入、音讯属性长度超过 65536 个字符则回绝音讯写入、PageCache 忙则报错。而后调用 CommitLog.putMessage 存入音讯。
  • 看到这里应该略微相熟一些了，终于到咱们期待已久的 CommitLog 出场了。次要是提早音讯解决，而后获取能够写入的 CommitLog 进行写入。
    提早音讯解决：如果音讯的提早级别大于 0，将音讯的原主题名称与原音讯队列 ID 存入音讯属性中，用提早音讯主题 SCHEDULE_TOPIC、音讯队列 ID 更新原先音讯的主题与队列，这是并发音讯生产重试要害的一步。但不是这个本节的次要指标，后文会进一步剖析。
    关键点在如何获取能够写入的 CommitLog。存储构造大节外面有提到每个 CommitLog 默认大小 1G，写完一个文件，以偏移量命名创立下一个文件。每个 1G 大小 CommitLog 的在代码层面对应的是 MappedFile，而多个 MappedFiled 组成 MappedFileQueue。逻辑上的 CommitLog 通过持有 MappedFileQueue 治理多个 MappedFile。所以，获取能够写入的 CommitLog 也就是获取 MappedFileQueue 最初一个 MappedFile，为什么是最初一个，因为后面的曾经写完了呀。来看看 RocketMQ 逻辑与物理存储的对应关系应该可能更直观的了解。
  • 获取到最初一个 MappedFile 后，调用 MappedFile.appendMessage 将音讯追加到该文件中。可是只管是程序写入，然而连小学生都晓得写磁盘还是很慢，难道想这样撑持 RocketMQ 单机十万吞吐量？too young too simple！从逻辑存储构造和物理存储构造的映射关系来看，MappedFile 持有物理 CommitLog 的 fileChannel (Java NIO 文件读写的通道)，通过 fileChannel 能够拜访物理 CommitLog 文件，然而 RocketMQ 并没有间接应用 fileChannel，而是映射到一个 MappedByteBuffer，咱们的目标就是把音讯写入这个 ByteBuffer 中，进而写入 MappedFile 对应的 CommitLog 文件。为什么须要这样做，还有哪些细节，会在”文件内存映射“小结中为大家解答。

  • 持续看流程，失去 MappedFile 对应的 ByteBuffer，咱们须要将音讯序列化，写入 ByteBuffer 中。

    1. 构建音讯 id, createMessageId
    2. 获取该音讯在音讯队列的偏移量，CommitLog 中保留了以后所有音讯队列的以后待写入偏移量。
    3. 判断是否是事务音讯：这里次要解决 Prepared 类型和 Rollback 类型的音讯，设置音讯 queueOffset 为 0
    4. 计算音讯总大小，calMsgLength。
    5. 判断文件的残余空间，是否足够写入当条音讯，如果不能够，则将文件开端写入残余空间大小 + 固定魔数；而后返回一个 END_OF_FILE 的后果
    6. 如果空间足够，这将这条音讯写入之前失去的 MappedFile 的 ByteBuffer 中。
    7. 将各字段依照”音讯在 CommitLog 各字段示意图“存入 Bytebuffer，而后返回 PUT_OK 后果

    总结

    以上就是明天 RocketMQ 音讯存储的次要内容，音讯只是写入到 CommitLog 对应的 ByteBuffer 中，下一期就是咱们重要的零拷贝行将退场。咱们简略总结一下明天的内容：

  • 要了解音讯的存储流程须要先晓得音讯的存储构造：在物理上音讯挨个程序写入 CommitLog，为了晋升音讯查问效率须要构建音讯的索引文件 ConsumeQueue/IndexFile；
  • Broker 启动时进行参数解析，并初始化了 NettyRemotingServer，启动存储服务用于音讯存储及索引构建等；
  • Broker 收到音讯存储申请，通过层层校验，获取 CommitLog 对应的 MappedFile，将音讯写入 MappedFile 对应的内存映射 ByteBuffer；

以上就是明天全副的内容，如果感觉本期的内容对你有用的话记得点赞、关注、转发、珍藏，这将是对我最大的反对。

如果你须要 RocketMQ 相干的所有材料，能够评论区留言，或者关注公众号：三此君。回复：mq，即可。

音讯曾经写入 ByteBuffer，写入 ByteBuffer 就能够了吗？那收到音讯间接抛弃岂不是更好。音讯要落在磁盘上才不会失落，所以下一期咱们要分享的就是音讯的刷盘及索引构建，PageCache 及零拷贝也将闪亮退场。感激观看，下期不见不散。

参考文献

  • RocketMQ 官网文档
  • RocketMQ 源码
  • 丁威, 周继锋. RocketMQ 技术底细：RocketMQ 架构设计与实现原理. 机械工业出版社, 2019-01.
  • 李伟. RocketMQ 分布式消息中间件：外围原理与最佳实际. 电子工业出版社, 2020-08.
  • 杨开元. RocketMQ 实战与原理解析. 机械工业出版社, 2018-06.
退出移动版