本文作者:刘涛,阿里云智能技术专家。
01 Compaction Topic介绍
一般来说,消息队列提供的数据过期机制有如下几种,比如有基于时间的过期机制——数据保存多长时间后即进行清理,也有基于数据总量的过期机制——数据分区数据量达到一定值后进行清理。
而 Compaction Topic 是一种基于 key 的数据过期机制,即对于相同 key 的数据只保留最新值。
该特性的应用场景主要为维护状态信息,或者在需要用到 KV 结构时,可以通过 Compaction Topic 将 key-value 信息直接保存到 MQ,从而解除对外部数据库的依赖。比如维护消费位点,可以将消费组加分区作为 key ,将消费位点做 offset ,以消息形式发送到 MQ ,压缩之后,消费时获取最新 offset 信息即可。另外,像 connect 里的 source 信息比如 Binlog 解析位点或其他 source 处理的位点信息均可存到 Compaction Topic。同时 Compaction Topic 也支持 存储 RSQLDB 与 RStreams 的 checkpoint 信息。
02 需要解决的问题
Compaction 过程中,需要解决如下几个问题:
第一,数据写入过程中,数据如何从生产者发送到 broker 并且最终落盘,数据主备之间的 HA 如何保证?
第二,整个 compaction 的流程包括哪几个步骤?如果数据量太大,如何优化?
第三,数据消费时如何索引消息?如果找不到消息指定的 offset 消息,如何处理?
第四,如果有机器故障,如何恢复老数据?
03 方案设计与实现
第一,数据如何写入。
首先写入到 CommitLog,主要为复用 CommitLog 本身的 HA 能力。然后通过 reput线程将 CommitLog 消息按照 Topic 加 partition 的维度拆分到不同文件里,按分区整理消息,同时生成索引。这样最终消息就按 Topic 加 partition的粒度做了规整。
在 compaction 过程中,为什么不在原先的 commitLog 上做规整,而是再额外按分区做规整?原因如下:
- 所有数据都会写到 CommitLog ,因此单个 Topic 的数据不连续。如果要遍历单个 topic 的所有数据,可能需要跳着读,这样就会导致大量冷读,对磁盘 IO 影响比较大。
- CommitLog 数据有自动过期机制,会将老数据删除,因此不能将数据直接写到 CommitLog,而 CompactionLog 里的老数据为按 key 过期,不一定会删除。
- compact 以分区为维度进行。如果多个分区同时做 compact ,效率较低。因为很多分区的 key 同时在一个结构里,会导致同一个分区能够 compact 的数据比较少, 并且 compact 之后也需要重新写一份么,因此,索性就在 compact 之前将消息通过 reput service 重新归整一遍。
Compact 流程如下:
第一步,确定需要做 compaction 的数据文件列表。一般大于两个文件,需要排除当前正在写的文件。
第二步,将上一步筛选出的文件做遍历,得到 key 到 offset 的映射关系。
第三步,根据映射关系将需要保留的数据重新写到新文件。
第四步,用新文件替换老文件,将老文件删除。
第二步的构建 OffsetMap 主要目的在于可以知道哪文件需要被保留、哪文件需要被删除,以及文件的前后关系,这样就可以确定写入的布局,确定布局之后,就可以按照append 的方式将需要保留的数据写到新文件。
此处记录的并非 key 到 value 的信息,而是 key 到 Offset 的信息。因为 value 的数据 body 可能较长,比较占空间,而 offset 是固定长度,且通过 offset 信息也可以明确消息的先后顺序。另外,key 的长度也不固定,直接在 map 存储原始 key 并不合适。因此我们将 MD5 作为新 key ,如果 MD5 相同 key 认为也相同。
做 compaction 时会遍历所有消息,将相同 key 且 offset 小于 OffsetMap 的值删除。最终通过原始数据与 map 结构得到压缩之后的数据文件。
上图为目录结构展示。写入时上面为数据文件,下面为索引,要 compact 的是标红两个文件。压缩后的文件存储于子目录,需要将老文件先标记为删除,将子目录文件与 CQ 同时移到老的根目录。注意,文件与 CQ 文件名一一对应,可以一起删除。
随着数据量越来越大,构建的 OffsetMap 也会越来越大,导致无法容纳。
因此不能使用全量构建方式,不能将所有要 compact 的文件的 OffsetMap 一次性构建,需要将全量构建改为增量构建,构建逻辑也会有小的变化。
第一轮构建:如上图,先构建上面部分的 OffsetMap ,然后遍历文件,如果 offset 小于 OffsetMap 中对应 key 的 offset 则删除,如果等于则保留。而下面部分的消息的offset 肯定大于 OffsetMap 内的 offset ,因此也需要保留。
第二轮构建:从上一次结束的点开始构建。如果上一轮中的某个 key 在新一轮中不存在,则保留上一轮的值;如果存在,则依然按照小于删除、大于保留的原则进行构建。
将一轮构建变为两轮构建后, OffsetMap 的大小显著降低,构建的数据量也显著降低。
原先的索引为 CommitLog Position、Message Size 和 Tag Hush,而现在我们复用了bcq 结构。由于 Compact 之后数据不连续,无法按照先前的方式直接查找数据所在物理位置。由于 queueOffset 依然为单调增排列,因此可以通过二分查找方式将索引找出。
二分查找需要 queueoffset 信息,索引结构也会发生变化,而 bcq 带有 queueoffse 信息,因此可以复用 bcq 的结构。
Queueoffset 在 compact 前后保持不变。如果 queueoffset 不存在,则获取第一个大于 queueoffset 的消息,然后从头开始将所有全量数据发送给客户端。
机器故障导致消息丢失时,需要做备机的重建。因为 CommitLog 只能恢复最新数据,而 CompactionLog 需要老数据。之前的 HA 方式下,数据文件可能在 compact 过程中被被删除,因此也不能基于复制文件的方式做主备间同步。
因此,我们实现了基于 message 的复制。即模拟消费请求从 master 上拉取消息。拉取位点一般从 0 开始,大于等于 commitLog 最小offset 时结束。拉取结束之后,再做一次 force compaction 将 CommitLog 数据与恢复时的数据做一次 compaction ,以保证保留的数据是被压缩之后的数据。后续流程不变。
04 使用说明
生产者侧使用现有生产者接口,因为要按分区做 compact ,因此需要将相同 key 路由到相同的 MessageQueue,需要自己实现相关算法。
消费者侧使用现有消费者接口,消费到消息后,存入本地类 Map 结构中再进行使用。我们的场景大多为从头开始拉数据,因此需要在开始时将消费位点重置到0。拉取完以后,将消息 key 与 value 传入本地 kv 结构,使用时直接从该结构拿取即可。