theme: smartblue

本文正在参加「金石计划 . 瓜分6万现金大奖」

RocketMQ的发展历史

RocketMQ是一个统一消息引擎、轻量级数据处理平台。RocketMQ是一款阿里巴巴开源的消息中间件。 2016 年 11 月 28 日，阿里巴巴向 广西党性培训 Apache 软件基金会捐赠RocketMQ，成为 Apache 孵化项目。 2017 年 9 月 25 日，Apache 宣布 RocketMQ孵化成为 Apache 顶级项目（TLP ），成为国内首个互联网中间件在 Apache 上的顶级项目。

RocketMQ的定位说明

RocketMQ作为一款基于磁盘存储的中间件，具有无限积压能力，并提供高吞吐、低延迟的服务能力，其最核心的部分必然是它优雅的存储设计。本系列文章主要针对于RocketMQ的多个关键特性的实现原理进行深入介绍，并对消息中间件遇到的各种问题进行总结，阐述 RocketMQ如何解决这些问题。

RocketMQ的核心工作机制

由上图可以看到RocketMQ存储的文件主要包括Commitlog文件、ConsumeQueue文件、Index文件。而对于消息存储是RocketMQ中最为复杂和最为重要的一部分，接下来会从RocketMQ的消息存储整体架构、PageCache与Mmap内存映射以及RocketMQ中两种不同的刷盘方式三方面来分别展开叙述。

RocketMQ的存储设计介绍

• CommitLog：RocketMQ将所有主题的消息存储在同一个文件中，确保消息发送时按顺序写文件，尽最大能力确保消息发送的高可用性与高吞吐量。

• ConsumeQueue：消息中间件一般都是基于Topic的订阅与发布模式，消息消费时必须按照主题进行筛选消息，显然从Commitlog文件中按照topic去筛选消息会变得及其低效，为了提高根据主题检索消息的效率，RocketMQ引入了ConsumeQueue文件，俗成消费队列文件。

• index文件：关系型数据库可以按照字段属性进行记录检索，作为一款主要面向业务开发的消息中间件，RocketMQ也提供了基于消息属性的检索能力，底层的核心设计理念是为Commitlog文件建立哈希索引，并存储在Index文件中。

在RocketMQ中顺序写入到Commitlog文件后，ConsumeQueue与Index文件都是异步构建的，其数据流向图如下：

如果觉得上面的流程过于复杂的话，那么我就就给大家展示一个最简单的模式图：

根据上面的图可以看出来了吧。就是三者之间的关系和联系，那么我们来仔细以下这三个介质的的底层实现是什么。

Commitlog文件

消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G 。

CommitLog的命名规则

• 从上面的图中也可以看得出来，其文件的命名也及其巧妙，使用该存储在消息文件中的第一个全局偏移量来命名文件文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；消息主要是顺序写入日志文件，当文件满了，写入下一个文件；这样的设计主要是方便根据消息的物理偏移量，快速定位到消息所在的物理文件，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。

CommitLog的性能读写

RocketMQ在消息写入过程中追求极致的磁盘的读写性能。所有主题的消息随着到达Broker的顺序写入commitlog文件。Commitlog文件使用顺序写，极大提高了文件的写性能。 当顺序依次追加到文件中，消息一旦写入，就无法再进行修改。Commitlog文件的具体布局如下图所示：

基于磁盘文件的与基于内存读取机制有一个本质的不同点，就是在内存读取模式下基本上是现成的数据结构，例如，数据、集合或者哈希表等，对数据的读写非常方便，但是针对于磁盘存储读取的Commitlog文件，我们该如何如何搜索，这时候我们引入了ConsumeQueue，我们后面会进行相关的说明和介绍。

• RocketMQ与关系型数据会为每一条数据引入一个ID主键，在基于磁盘的读取机制中，也会为一条Message引入一个唯一标志：消息物理偏移量，即消息存储在文件的起始位置。

• 正是有了物理偏移量的概念，这也与上面提到的Commitlog的文件名命名相互呼应，这样做的好处是给出任意一个消息的物理偏移量，例如消息偏移量为 12345678，可以通过二分法进行查找，快速定位这个文件在第一个文件中，然后用消息的物理偏移量减去该文件的名称所得到的差值，就是在该文件中的绝对地址。

ConsumeQueue的读取模式

ConsumeQueue消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。

• ConsumeQueue文件是Commitlog文件基于Topic的索引文件，主要用于消费者根据Topic消费消息，其组织方式为/topic/queue，同一个队列中存在多个文件。

• Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。

注意：ConsumeQueue中的每个条目长度固定20个字节（8字节commitlog物理偏移量、4字节消息长度、8字节tag hashcode（这里不是存储tag的原始字符串，而选择存储hashcode）），每个条目的长度固定，可以使用访问类似数组下标的方式快速定位条目，极大地提高了ConsumeQueue文件的读取性能。

ConsumeQueue的读取消息偏移量

1. 首先，消息消费者根据topic、消息消费进度（ConsumeQueue逻辑偏移量），即第几个ConsumeQueue条目，这样的消费进度去访问消息的方法为使用逻辑偏移量logicOffset * 20即可找到该条目的起始偏移量（ConsumeQueue文件中的偏移量），然后读取该偏移量后20个字节即得到一个条目，无须遍历ConsumeQueue文件。

• 第N个消ConsumeQueue的元素数据的索引开始：（N-1）* 20+1
• 第N个消ConsumeQueue的元素数据的索引结束：（N）* 20

ConsumeQueue文件夹如下：topic/queue/file三层组织结构，地址：$HOME/store/ConsumeQueue/{topic}/{queueId}/{fileName}。单个ConsumerQueue文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；

存储结构分析总结

• 数据单独存储到一个Commit Log完全
• 队列实际只存储消息在Commit Log的位置信息，并且串行方式刷盘。

这样做的优点

• 队列轻量化，单个队列数据量非常少。
• 对磁盘的访问串行化，避免磁盘竟争，不会因为队列增加导致 IOWAIT 增高。

这样做的缺点

• 写虽然完全是顺序写，但是读却变成了完全的随机读。
• 读一条消息，会先读 Consume Queue，再读 Commit Log，增加了开销。
• 要保证 Commit Log 与 Consume Queue 完全的一致，增加了编程的复杂度。

以上缺点如何克服

• 随机读，尽可能让读命中PAGECACHE，减少 IO 读操作，所以内存越大越好。

• 访问 PAGECACHE 时，即使只访问1k 的消息，系统也会提前预读出更多数据，在下次读时，就可能命中内存。

• 随机访问 Commit Log 磁盘数据，系统 IO 调度算法设置为 NOOP 方式，会在一定程度上将完全的随机读变成顺序跳跃方式，而顺序跳跃方式读较完全的随机读性能会高5倍以上。

4k的消息在完全随机访问情况下，仍然可以达到8K次每秒以上的读性能。RocketMQ与Kafka相比具有一个强大的优势，就是支持按消息属性检索消息，引入consumequeue文件解决了基于topic查找的问题，但如果想基于消息的某一个属性查找消息，ConsumeQueue文件就无能为力了，为了解决此问题RocketMQ引入了Index索引文件，实现基于文件的哈希索引。

ConsumeQueue和CommitLog的总结介绍

CommitLog 中存储了所有的元信息，包含消息体，类似于 Mysql、Oracle 的 binlog，所以只要有CommitLog 在，ConsumeQueue即使数据丢失，仍然可以恢复出来。

IndexFile文件

IndexFile文件基于物理磁盘文件实现Hash索引。其文件由40字节的文件头、500万个哈希槽，每个哈希槽4个字节，最后由2000万个Index条目，每个条目由20个字节构成，分别为4字节索引key的hashcode、8字节消息物理偏移量、4字节时间戳、4字节的前一个Index条目（哈希冲突的链表结构）。

IndexFile的文件存储结构如下图所示：

其原理和读取方式与ConsumerQueue较为相似，至此不过多赘述。对于IndexFile文件和ConsumerQueue文件都是，Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。

页缓存与内存映射

RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在PageCache机制的预读取作用下，ConsumeQueue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。（此时块存储采用SSD的话），随机读的性能也会有所提升。

内存映射mmap

虽然基于磁盘的顺序写可以极大提高IO的写效率，但如果基于文件的存储采用常规的JAVA文件操作API，例如 FileOutputStream等，其性能提升会很有限，RocketMQ引入了内存映射，将磁盘文件映射到内存中，以操作内存的方式操作磁盘，性能又提升了一个档次。

JAVA中可通过FileChannel的map方法创建内存映射文件。在Linux服务器中由该方法创建的文件使用的是操作系统的pagecache，即页缓存。

• Linux操作系统中的内存使用策略时会尽可能地利用机器的物理内存，并常驻内存中，就是所谓的页缓存。在操作系统的内存不够的情况下，采用缓存置换算法，例如LRU将不常用的页缓存回收，即操作系统会自动管理这部分内存。

• 如果RocketMQ Broker进程异常退出，存储在页缓存中的数据并不会丢失，操作系统会定时将页缓存中的数据持久化到磁盘，做到数据安全可靠。不过如果是机器断电等异常情况，存储在页缓存中的数据就有可能丢失。

页缓存PageCache

页缓存（PageCache)是OS对文件的缓存，用于加速对文件的读写。

• 程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。

  • 对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。
  • 对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。

是否使用堆外内存

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

默认情况下RocketMQ将消息写入PageCache，消息消费时从PageCache中读取，这样在高并发时PageCache的压力会比较大，容易出现瞬时broker busy。

RocketMQ还引入了transientStorePoolEnable=true，将消息先写入堆外内存并立即返回，然后异步将堆外内存中的数据提交到pagecache，再异步刷盘到磁盘中。其工作机制如下图所示：

消息在消费读取时不会尝试从堆外内存中读，而是从PageCache中读取，这样就形成了内存级别的读写分离，即消息写入时主要面对堆外内存，而读消息时主要面对pagecache。

• 优点是消息是直接写入堆外内存，然后异步写入pagecache。相比每条消息追加直接写入pagechae，其最大的优势是将消息写入PageCache操作批量化。

• 缺点是如果由于某些意外操作导致Broker进程异常退出，那么存储在堆外内存的数据会丢失，但如果是放入pagecache，broker异常退出并不会丢失消息。

刷盘机制

有了顺序写和内存映射的加持，RocketMQ的写入性能得到了极大的保证，但凡事都有利弊，引入了内存映射和页缓存机制，消息会先写入到页缓存，此时消息并没有真正持久化到磁盘。那么broker收到客户端的消息发送后，是存储到页缓存中就直接返回成功，还是要持久化到磁盘中才返回成功呢？

这是一个“艰难”的抉择，是在性能与消息可靠性方面进行权衡。为此，RocketMQ提供了多种策略：同步刷盘、异步刷盘。

同步刷盘

同步刷盘在RocketMQ的实现中成为组提交，并不是每一条消息都必须刷盘。采用同步刷盘，每一个线程将数据追到到内存后，并向刷盘线程提交刷盘请求，然后会阻塞；刷盘线程从任务队列中获取一个任务，然后触发一次刷盘，但并不只刷与请求相关的消息，而是会直接将内存中待刷盘的所有消息一次批量刷盘，然后就可以唤醒一组请求线程，实现组刷盘。

同步刷盘的优点是能保证消息不丢失，即向客户端返回成功就代表这条消息已被持久化到磁盘，即消息非常可靠，但这是以牺牲写入响应延迟性能为代价的，由于RocketMQ的消息是先写入pagecache，故消息丢失的可能性较小，如果能容忍一定几率的消息丢失，可以考虑使用异步刷盘。

如上图所示，只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多。

异步刷盘

异步刷盘指的是broker将消息存储到pagecache后就立即返回成功，然后开启一个异步线程定时执行FileChannel的forece方法，将内存中的数据定时刷写到磁盘，默认间隔为500ms。

能够充分利用OS的PageCache的优势，只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了MQ的性能和吞吐量。

GroupCommitService 从队列中拿出待刷盘请求request， 然后执行刷盘动作， 此时会将write指针与flush指针之间的所有数据刷写到磁盘中，即这里并不只是将request l对应的那一条消息刷写到磁盘

MySQL的持久化机制对比

MySQL Redo 日志的引入目的，我们知道在 MySQL InnoDB 的存储引擎中，会有一个内存 Pool，用来缓存磁盘的文件块，当更新语句将数据修改后，会首先在内存中进行修改，然后将变更写入到 redo 文件(刷写到磁盘)，然后定时将InnoDB内存池中的数据刷写到磁盘。

RocketMQ的持久化存储机制总结

RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)，针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构。

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

• Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后，可以等下一次消息拉取，同时服务端也支持长轮询模式，如果一个消息拉取请求未拉取到消息，Broker允许等待30s的时间，只要这段时间内有新消息到达，将直接返回给消费端。