基本概念
参考官方: https://github.com/apache/rocketmq/tree/master/docs/cn
整体架构
- 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请求放入列表,并阻塞等待。后台线程从列表中获取请求并刷新磁盘,成功刷盘后通知等待线程。
RocketMQ总结之 消息发送和存储流程
原文:https://www.cnblogs.com/ft-greate/p/12880852.html