图解RocketMQ消息发送和存储流程

基本概念

• 参考官网文档

整体架构

• Producer：生产者
• Consumer：消费者
• Broker：负责消息存储、投递、查询
• NameServer：路由注册中心。功能包括：Broker管理、路由信息管理

模块间数据流转

生产-消费模型

消息发送流程

• Broker启动时，向NameServer注册信息
• 客户端调用producer发送消息时，会先从NameServer获取该topic的路由信息。消息头code为GET_ROUTEINFO_BY_TOPIC
• 从NameServer返回的路由信息，包括topic包含的队列列表和broker列表
• Producer端根据查询策略，选出其中一个队列，用于后续存储消息
• 每条消息会生成一个唯一id，添加到消息的属性中。属性的key为UNIQ_KEY
• 对消息做一些特殊处理，比如：超过4M会对消息进行压缩
• producer向Broker发送rpc请求，将消息保存到broker端。消息头的code为SEND_MESSAGE或SEND_MESSAGE_V2（配置文件设置了特殊标志）

消息存储流程

• Broker端收到消息后，将消息原始信息保存在CommitLog文件对应的MappedFile中，然后异步刷新到磁盘
• ReputMessageServie线程异步的将CommitLog中MappedFile中的消息保存到ConsumerQueue和IndexFile中
• ConsumerQueue和IndexFile只是原始文件的索引信息

消息体结构

• CommitLog的消息体长度不一样，每个CommitLog文件默认1G
• ConsumerQueue内的消息体长度固定，为20Byte

内存映射流程

• 内存映射文件MappedFile通过AllocateMappedFileService创建
• MappedFile的创建是典型的生产者-消费者模型
• MappedFileQueue调用getLastMappedFile获取MappedFile时，将请求放入队列中
• AllocateMappedFileService线程持续监听队列，队列有请求时，创建出MappedFile对象
• 最后将MappedFile对象预热，底层调用force方法和mlock方法

刷盘流程

• producer发送给broker的消息保存在MappedFile中，然后通过刷盘机制同步到磁盘中
• 刷盘分为同步刷盘和异步刷盘
• 异步刷盘后台线程按一定时间间隔执行
• 同步刷盘也是生产者-消费者模型。broker保存消息到MappedFile后，创建GroupCommitRequest请求放入列表，并阻塞等待。后台线程从列表中获取请求并刷新磁盘，成功刷盘后通知等待线程。