目录一、RocketMQ消息是如何存储的二、RocketMQ延迟消息是如何实现的三、RocketMQ提供了哪些消息过滤机制一、RocketMQ消息是如何存储的RocketMQ的消息存储是一个复杂而高效的过程设计上充分考虑了性能和扩展性。消息存储的主要组件包括CommitLog文件、消费队列文件(ConsumeQueue)、以及索引|文件(IndexFile)。我们来详细解释一下这几个核心部分。1. CommitLog文件CommitLog是RocketMQ的核心存储文件负责保存消息的完整内容。顺序写入所有的消息都顺序写入Commitlog文件这种方式减少了磁盘寻道时间提高了写入性能。文件滚动CommitLog按照固定大小(比如1GB)进行分片。当一个文件写满后会创建一个新的文件。存储所有数据包括消息体、主题、队列ID等。2. ConsumeQueue文件ConsumeQueue是针对消息的逻辑视图旨在加快消费者对消息的访问速度。条目固定每个ConsumeQueue条目固定为20字节包含消息在CommitLog中的偏移量、消息大小、Tag哈希值。独立文件:每个主题的每个队列都有独立的ConsumeQueue文件文件路径为store/consumequeue/{topic}/{queueId}。快速定位通过ConsumeQueue消费者无需扫描整个CommitLog即可快速找到消息的位置。3.IndexFile文件IndexFile用于支持消息的快速检索。哈希索引为消息的key建立哈希索引支持通过key快速检索消息偏移。增强查询IndexFile是可选的用于需要基于消息属性进行快速查找的场景。消息存储流程1.接收消息:Broker接收到消息后将其放入内存缓冲区(待写入CommitLog)。2.写入CommitLog:每条消息追加到当前活跃的CommitLog文件中。使用顺序写入提升写入效率和磁盘利用率。3.同步到ConsumeQueue:异步转发服务(ReputMessageService)从CommitLog读取新写入的消息。将消息的偏移量和其他元数据(如大小和Tag哈希值)存储到相应的ConsumeQueue文件中。4.更新IndexFile(可选)若消息带有key(如业务ID)则将其哈希和偏移量存入IndexFile。这样可以通过该key快速查找消息。二、RocketMQ延迟消息是如何实现的RocketMQ通过特定的延迟级别设计实现延迟消息功能。在RocketMQ中延迟消息是通过设置消息的延迟级别(DelayLevel)来实现的。每个延迟级别对应一个特定的时间段这样可以让消息在指定的时间之后才被消费。实现原理1. 延迟级别:RocketMQ不支持任意时间的延迟而是提供了18个固定的延迟级别从1s5s10s30s1m2m,3m到2h不等。2.特殊主题所有延迟消息都会先发送到一个特殊的内部主题SCHEDULE_TOPIC_XXXX。3.定时任务Broker会启动一个定时任务按照延迟时间的先后顺序依次扫描每个延迟级别队列。4.消息转移当扫描到期的消息时会将消息从延迟队列转移到目标主题。5.消费者消费消息被转移到目标主题后消费者就可以正常消费这条消息了。三、RocketMQ提供了哪些消息过滤机制RocketMQ提供了多种消息过滤机制以便消费者能够根据业务需求进行精确的消息消费。主要的过滤机制包括1.Tag过滤最常用的过滤方式消费者可以基于消息的Tag进行过滤。2.SQL92语法过滤基于Message属性以SQL92标准语法进行复杂条件过滤该功能需要Broker支持。