xijiu

一、概述

“索引”一种数据结构,帮助我们快速定位、查询数据

前文我们梳理了消息在Commit Log文件的存储过程,讨论了消息的落盘策略,然而仅仅通过Commit Log存储消息是远远不够的,例如当我们需要消费某个topic的消息时,通过对Commit Log整体遍历寻找消息的方式无疑非常的低效。所以本章将引出2个很重要的概念:消费队列IndexFile

二、消费队列

2.1 概念

什么是消费队列呢?其实在上一章的消息协议格式中,就有消息队列的体现,简单回顾一下协议的前6个字段:

  • msg total len:消息总长度
  • msg magic   :魔法值,标记当前数据是一条消息
  • msg CRC     :消息内容的CRC值
  • queue id      :队列id
  • msg flag      :消息标记位
  • queue offset :队列的偏移量,从0开始累加

其中第4个字段为消费队列的id,第6个字段为当前队列的偏移量;所以在消息产生的时候,消息所属的队列就已经确定

那么究竟该如何理解“消费队列”的概念呢?我们举例来说:假定某个RocketMQ集群部署了3个broker(brokerA、brokerB、brokerC),主题topicTest的消息分别存储在这3个broker中,而每个broker又对应一个commit log文件。我们把broker中的主题topicTest中的消息划分为多个队列,每个队列便是这个topic在当前broker的消费队列

2.2 数据结构

当然消费队列不会将消息体进行冗余存储,数据结构如下:

ConsumeQueue数据结构

即在消费队列的文件中,需要存储20个字节的索引内容。RocketMQ中默认指定每个消费队列的文件存储30万条消息的索引,而一个索引占用20个字节,这样每个文件的大小便是固定值300000*20/1024/1024≈5.72M,而文件命名采用与commit log相似的方式,即总长度20位,高位补0

  • store/consumequeue/topicXX/0/00000000000000000000 第一个文件
  • store/consumequeue/topicXX/1/00000000000006000000 第二个文件
  • store/consumequeue/topicXX/2/00000000000012000000 第三个文件
  • store/consumequeue/topicXX/3/00000000000018000000 第四个文件

与commitLog只有一个文件不同,consumeQueue是每个topic的每个消费队列都会生产多个文件

为什么消费队列文件存储消息的个数要设置成30万呢?一个文件还不到6M,为何不能像commit log那样设置为1G呢?鄙人没有在源码及网上找到相关资料,猜测可能是个经验值。首先该值不宜设置的过大,因为消息总是有失效期的,例如3天失效,如果消费队列的文件设置过大的话,有可能一个文件中包含了过去一个月的消息,时间跨度过大,这样不利于及时删除已经过期的消息;其次该值也不宜过小,太小的话会产生大量的小文件,在管理维护上制造负担。最理想情况是一个消费队列文件对应一个commit log,这样commit log过期时,消费队列文件也跟着及时失效

2.3 消费队列之commit log视角

CommitLog视角中的ConsumeQueue文件

某个commit log会存储多个topic消息,而每个topic有可能会将消息划分至多个队列中;如上图所示,commit log按顺序依次存储消息,而某个topic的消息在commit log中大概率也是不连续的,而consume queue的作用便是将某个topic下同一个队列的消息依次标识,便于消费时顺序消费

2.4 消费队列之topic视角

topic视角中的ConsumeQueue文件

上图描述了Topic A的消费队列分配情况,所有的消息相对均匀的分散在3个broker中,每个broker的消息又分为队列0及队列1,所以一共有6个消费队列,所以Topic A的consumer端也是建议开辟6个进程去消费数据

这里简单提一下消费端,我们知道一个消费队列同时只能被一个消费实例消费,所以消费实例的数量建议值为 <= consumeQueue 数量,理想情况是消费实例的个数完全等于consumeQueue个数,这样吞吐能达到最佳,以下:

  1. consumerNum < consumeQueue 消费实例小于消费队列个数。例如某个topic的消费队列一共有6个,但是只有3个消费实例,RocketMQ会尽量均衡每个消费实例分配到的消费队列,所以每个消费实例实际会消费2个队列的内容。这种情况可能增加消费实例可以提高整体吞吐
  2. consumerNum > consumeQueue 消费实例大于消费队列个数。比如某个topic的消费队列有6个,但是有8个消费实例注册,因为一个消费实例只能对应一个消费队列,所以势必导致有2个消费实例处于空闲状态,不会拿到任何数据
  3. consumerNum == consumeQueue 消费实例等于消费队列个数。这比较理想的状态,不会有过载或饥饿产生

基于同一个消费队列只能被一个消费实例消费的特性,我们可以将某类消息均发送给一个队列,这样消费的时候能够严格保序。例如我们希望订单的流程是保序的,可以通过orderId % consumeQueue来决定当前订单的消费发送给哪个队列,从而达到保序的目的

2.5 写入流程

消费队列的写入跟commit log的写入是同步进行的吗?答案是否定的,RocketMQ会启动一个独立的线程来异步构建消费队列(构建索引文件也是这个线程)

ConsumeQueue文件构建

简单描述下流程:构建索引的线程为ReputMessageService,跟写入commitLog的线程是异步关系,该线程会不断地将没有构建索引的消息从commit log中取出,将物理偏移量、消息长度、tag写入文件。值得一提的是,消息队列文件的写入跟commit log不同,commit log的写入有很多刷盘策略,而consumeQueue每条消息解析完毕都会刷盘,而且采用的是FileChannel

借此,我们抛出几个问题

问题1:为什么消费队列写入文件要用FileChannel?批次多,数据量小的场景用Mapp不香吗?

关于这个问题,我咨询了RocketMQ开源社区比较有影响力的大佬,给出的答复是:的确是这样,RocketMQ这样设计考虑欠佳,写文件这块应该向kafka学习,即消息写入用FileChannel,索引写入用Mapp

问题2:为什么RocketMQ中很少有用到堆外内存?文件写入的话,使用堆外内存少一次内存拷贝,不是可以提高性能吗?

是这样的,类似这样的场景首选还是堆外内存;RocketMQ的确还有很多可优化的空间,在将来的某个版本,我们一定可以看到针对此处的优化

问题3:如果消息已经写入commit log,但还未写入消费队列,consumer端能正常消费到这条消息吗?

抛出这个问题,大家可以思考一下,在消息产生、消费的章节再回答

至此,消费队列在broker端的存储生命周期便结束了,不过在后文的消息生产、消费环节我们还会反复提及。虽然消费队列在文件操作时并不复杂,也没有像commit log那样复杂的刷盘策略,但我们需要深度理解topic、commit log、消费队列的概念及其从属关系,这样在后文介绍消息生产、消费时才能游刃有余

三、IndexFile

ReputMessageService线程除了构建消费队列的索引外,还同时根据消息key构建了索引

IndexFile文件构建

3.1 IndexFile简介

除了正常的生产、消费消息外,RocketMQ还提供了根据msg key进行查询的功能,将消息key相同的消息一并查出;我们当然可以通过扫描全量的commit log将相同msg key类型的消息过滤出来,但性能堪忧,而且涉及大量的IO运算;IndexFile便是为了实现快速查找目标消息而衍生的索引文件

IndexFile的命名规范也有别于消费队列,IndexFile是按照创建时间来命名的,因为根据消息key进行匹配查询的时候,都要带上时间参数,文件名起到了快速定位索引数据位置的作用,下面列举一组IndexFile的文件名

  • rocketMQ/store/20211204094647480
  • rocketMQ/store/20211205094647480
  • rocketMQ/store/20211206094647480
  • rocketMQ/store/20211207094647480

3.2 IndexFile结构

我们具体看一下此文件的结构,与消费队列文件相同,IndexFile是定长

IndexFile文件结构

由三部分组成:

  1. 文件头,占用 40 byte
  2. slot,hash槽儿,占用500w*4= 20000000 byte
  3. 索引内容 占用2000w*20= 400000000 byte

所以文件总大小为: 40+5000000*4+20000000*20=420000040byte ≈ 400M

3.3 存储原理

简单剖析一下各部分的字段

文件头 共 20 byte

  • 开始时间(8 byte)存储前索引文件内,所有消息的最小时间
  • 结束时间(8 byte)存储前索引文件内,所有消息的最大时间,因为根据key查询的时候 ,时间是必填选项 ,开始与结束时间用来快速定位消息是否命中
  • 最小物理偏移量(8 byte)存储前索引文件内,所有消息的最小物理偏移量
  • 最大物理偏移量(8 byte)存储前索引文件内,所有消息的最大物理偏移量;框定最小、最大物理偏移量,是为了给通过物理地址查询时快速索引
  • 有效hash slot数量(4 byte)因为存储hash冲突的情况,所以最坏情况是,hash slot只有1个,最理想情况是有500万个
  • index索引数量(4 byte)如果当前索引文件已经构建完毕,那么该值是固定值2000万

slot 4 byte

  • 当前槽儿内的最近一次index的位置(4 byte

索引内容 20 byte

  • hash值(4 byte
  • 消息的物理地址(8 byte
  • 时间差(4 byte)当前消息与最早消息的时间差
  • 索引(4 byte)当前槽儿内,上一条索引的位置

存储方式如下

IndexFile文件存储原理

当一条新的消息索引进来时,首先定位当前消息命中的slot,该slot存储着最近一条消息的存储位置。将消息的索引数据append至文件尾部的同时,将最新索引数据的next指向上一条索引位置,这样便形成了一条当前slot按照时间存入的倒序的链表

3.4 消息查询

根据前文的铺垫,同一个槽儿内的数据,已经被一个隐式的链儿串连在了一起,当我们根据topic+key进行数据查询时,直接通过topic + # + key的hash值,定位到某个槽儿,进而依次寻找消息即可;当然同一个槽儿内的数据可能出现hash冲突,我们需要将不符合条件的数据过滤掉

当我阅读这部分源码的时候,发现了其内部存在的一个bug,其做消息过滤时,仅仅判断消息的hash字段是否相等,如果相等的话,继而认定为要寻找的数据从而返回

class : org.apache.rocketmq.store.index.IndexFile

if (keyHash == keyHashRead && timeMatched) {
    phyOffsets.add(phyOffsetRead);
}

进而带来的一些问题,例如:

  • 新建topic AaTopic,并向topic中发送一条消息,message key为Aa
  • 新建topic BBTopic,并向topic中发送一条消息,message key为BB

当我们通过 topic=AaTopic && key=BB查询时,预期应该返回空数据,但实际却返回了一条数据

msgKey查询bug

其主要是因为AaBB拥有相同的HashCode2080

此bug以及解决办法已经在GitHub上做了提交:

3.5 message id

消息id,在RocketMQ中又定义为msg unique id,组成形式是“ip+物理偏移量”(ip非定长字段,会因ipv4与ipv6的不同而有所区别),其中ip及物理偏移量在消息的协议格式中均有体现;当我们拿到消息所属的broker地址,以及该消息的物理存储偏移量时,也就唯一定位了该条消息,所以使用“ip+物理偏移量”的方式作为消息id

在某些场景下,msg unique id也会存储在indexFile中,不是本文的重点,我们后文还会提及

四、索引查询(page cache)

查询这块,我们将放在消息发送、消费的章节来阐述,此处仅仅讨论索引结构设计中page cache所承担的角色

其实在整个流程中,RocketMQ是极度依赖page cache的,尤其是消费队列,数据查询是通过如下流程来查询消息的:

1、broker接受请求 -> 2、查询ConsumeQueue文件(20byte) -> 3、拿到消息的physicOffset -> 4、查询commitLog文件(msg size)

我们发现第二步及第四步都只是查询很小的数据量,如果没有page cache挡在磁盘前,整体的性能必将是断崖式下降。我有朋友做过禁掉page cache后,RocketMQ前后的性能的对比相差好几个量级,不禁感慨,page cache真是让我们又爱又恨,叹叹

分类:

架构

技术点:

相关文章: