RocketMQ（二）之消息存储模块

2024-03-28 本文已影响0人 b77f686a8725a

本文涉及：

前言

CommitLog在RocketMQ中是消息存储的核心模块，它是RocketMQ中最主要的消息存储文件，负责存储所有消息的原始内容。每个Broker节点都会维护一个CommitLog文件，所有消息不管属于哪个Topic或Queue，都会按照消息的发送时间顺序，追加写入到CommitLog中。

追加写入：所有消息不分Topic和Queue，按照发送顺序依次追加到CommitLog文件中。这种方式可以充分利用磁盘的顺序写特性，提高写入性能并减少磁盘碎片。
内存映射：RocketMQ采用内存映射文件（MappedByteBuffer）技术，将CommitLog文件映射到内存，从而减少系统调用，提高写入效率。
刷盘策略：RocketMQ支持同步刷盘（FlushDiskType.SYNC_FLUSH）和异步刷盘（FlushDiskType.ASYNC_FLUSH）两种策略，以平衡消息持久化和性能。

ConsumeQueue：Consumer并不是直接从CommitLog读取消息，而是通过ConsumeQueue（消费队列）读取。ConsumeQueue是一个特殊的索引文件，记录了每个消息在CommitLog中的偏移量和消息大小，以此快速定位到CommitLog中的具体消息。
读取过程：当Consumer需要消费消息时，根据ConsumeQueue中的索引信息，直接从CommitLog中读取对应位置的消息内容。

过期删除：RocketMQ支持配置消息的存活时间（Time To Live, TTL），超过这个时间的消息将被视为过期消息，Broker会在后台进行垃圾回收，清理过期消息占用的CommitLog空间。
磁盘容量管理：Broker会根据预设的磁盘使用率阈值，触发磁盘空间检查。当磁盘使用率达到一定比例时，Broker会根据消息的优先级和过期时间删除最早的过期消息，释放磁盘空间。
Compact策略（可选）：在某些情况下，RocketMQ可以配置开启消息整理（Compaction）功能，对CommitLog进行压缩以节省空间，但这并非RocketMQ标准特性，也不是其主要回收策略。

ConsumeQueue 是一种特殊的索引文件，它的主要作用是为消息提供一种快速查找机制，使Consumer能迅速定位到实际存储消息内容的CommitLog 文件中的具体位置。

ConsumeQueue 是基于主题（Topic）和消费组（Consumer Group）组织的，每个Topic下的每个消费队列（Message Queue）都有一个独立的ConsumeQueue 文件。
ConsumeQueue 文件的物理结构采用定长设计，每个条目固定大小，方便快速随机读取。
文件名通常包含Topic、队列ID和文件序号，文件内部存储的是每条消息在CommitLog 中的物理偏移量（offset）、消息大小、消息的Tag Hashcode以及其他可能的元数据信息。

当Consumer需要消费消息时，它首先根据自己的订阅信息和消费进度从ConsumeQueue 文件中读取消息索引。消费者拿到索引后，通过索引中的offset值直接定位到CommitLog 文件的相应位置，从该位置开始读取消息内容。

在特定版本或场景下，可以开发消息索引机制，根据特定字段进行快速检索。

设计索引服务
- 定义索引模型：根据业务需求，确定需要对哪些RocketMQ消息字段进行索引，如Message Key、Tag、Header中的某些属性等。
- 选择索引库：选取适合的全文搜索引擎或者索引存储系统，如Elasticsearch、Solr、MongoDB甚至是自建索引数据库等。
- 设计索引结构：在索引库中定义相应的索引结构，包括但不限于字段类型、索引类型（如倒排索引）、分词策略等。
将提取出来的字段信息转换为适合全文检索引擎存储的格式。并将这些信息推送到Elasticsearch、Solr或其他类似的全文搜索引擎中，为其创建索引。
根据查询结果得到的消息Key（或者消息ID），再到RocketMQ中通过Message ID或者Key从CommitLog中精确地获取完整消息内容。