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

6,063 阅读2分钟

基本概念

整体架构

  • 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请求放入列表,并阻塞等待。后台线程从列表中获取请求并刷新磁盘,成功刷盘后通知等待线程。