本文涉及:
目录
前言
关于消息队列中间件
RocketMQ概述
RocketMQ(一)之基础架构
CommitLog
CommitLog在RocketMQ中是消息存储的核心模块,它是RocketMQ中最主要的消息存储文件,负责存储所有消息的原始内容。每个Broker节点都会维护一个CommitLog文件,所有消息不管属于哪个Topic或Queue,都会按照消息的发送时间顺序,追加写入到CommitLog中。
消息写入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
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中精确地获取完整消息内容。