1. 消息接收准备

当RocketMQ Broker接收到消息时，首先会对接收到的消息进行一些准备工作。这些准备工作包括：

  • 准备消费队列：Broker根据消息的Topic和队列数量创建或获取对应的消费队列。消费队列用于存储消息，并由消费者进行消费。
  • 更新消费进度：Broker会更新消费进度，将消息的偏移量追加到消费进度文件中。消费进度文件用于记录消费者消费的位置，以便在消费者宕机后能够继续消费未消费的消息。
  • 检查消息的合法性：Broker会对接收到的消息进行一些基础的验证，例如消息的大小、消息的是否为空等。
  • 生成消息索引：Broker会根据消息的属性和内容生成消息索引，以支持根据消息属性进行查询。

2. 消息持久化

在消息接收准备完毕后，接下来是进行消息的持久化。Broker将消息以文件的形式存储在磁盘上，以保证消息的可靠性和持久化。消息持久化过程包括：

  • 序列化：将消息对象序列化为字节数组，以便于存储和传输。
  • 创建物理文件：根据消息的Topic和队列ID，Broker创建对应的物理文件，用于存储消息。
  • 写入消息内容：将序列化后的字节数组写入到物理文件中，写入的位置是文件的末尾。
  • 更新消费队列信息：更新消费队列中的最大偏移量，标识当前队列中最新消息的偏移量。

3. 消息分发给消费者

当消息持久化完毕后，Broker会将消息分发给消费者进行消费。消息分发的过程如下：

  • 根据消费者订阅关系：Broker根据消费者的订阅关系，判断消息是否需要被该消费者消费。
  • 推送消息给消费者：如果消息需要被该消费者消费，Broker会将消息推送给消费者。推送方式有两种：同步推送和异步推送。
  • 同步推送：Broker等待该消费者消费完消息后返回消费结果，再继续向下一个消费者推送消息。
  • 异步推送：Broker将消息推送给消费者，不等待消费结果，继续向下一个消费者推送消息。

4. 消息删除

当消息被所有消费者消费完毕后，Broker会对消息进行删除。消息删除的过程如下：

  • 标记消息删除：Broker会将消费队列中的消息标记为删除状态。
  • 消息回收：定期或触发条件满足时，Broker会进行消息回收，将标记为删除的消息从磁盘上删除，释放存储空间。
  • 更新消费队列信息：更新消费队列中的最大偏移量，标识当前队列中最新消息的偏移量。