延迟消息

延迟等级

官方默认设置了 18 哥延迟等级

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

发送延迟消息：按照默认顺序 1-18 数字就对应上面的延迟时间

Message msg = new Message (TOPIC, TAG, "OrderID199", "ok", getBytes(StandardCharsets.UTF_8));
//设置延迟等级
msg.setDelayTimeLevel(3);
producer.send(msg);

基本原理

延迟消息都会被存储到 RocketMQ 的一个内部 Topic：SCHEDULE_TOPIC_XXXX 中

SCHEDULE_TOPIC_XXXX 共有 18 个 MessageQueue：

• 对应延迟消息的 18 个等级，根据指定的 DelayTimeLevel 来决定选择哪个 MessageQueue
• 有一个定时任务，每 100 ms 执行一次判断 SCHEDULE_TOPIC_XXXX Topic 中的 MessageQueue 的消息是否到达延迟时间
• 若到达延迟时间，将 SCHEDULE_TOPIC_XXXX 中的消息投递到消息最初需要投递的 Topic 之中

为什么不支持任意时间？

RocketMQ 并不支持任意时间的延迟，可能的主要原因就是如果提供任意时间，就会涉及到消息的排序，会有一定的性能损耗

事务消息

RocketMQ 采用了 2PC 的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或失败的消息

基本流程

第一阶段：

• 发送 Message，Half Message ，即半事务消息
• 此类型的 Message 是不会被 Consumer 消费的

第二阶段：如果半事务消息投递成功，则会开始执行本地事务

分为如下三种 Case：

• 本地事务执行成功：会为 Broker 发送 commit 消息，被 commit 过后的 Message 才能被 Consumer 消费
• 本地事务执行失败：
  • 会为 Broker 发送 rollback 消息，Broker 则会将刚刚投递的半事务消息删除，从而保证上下游数据的一致性
  • 如果 Producer 实例或者网络出现问题，Producer 没能及时 de 将本地事务执行的结果通知 Broker，Broker 会通过扫描发现某条 Message 长时间处于半事务状态，Broker 会主动 de 给 Producer 询问此 Message 对应的事务状态

基本设计

采用 2PC 两阶段设计：

将 Message 原本真实的 Topic 和 MessageQueue 进行备份

• 存入到 PROPERTY_REALTOPC、PROPERTY_REAL_QUEUE_ID 中

将消息投递到一个内部 Topic 中 RMQ_SYS_TRANS_HALF_TOPIC，该队列专门存储事务消息

所有的 Half Message 全部都写入到 queueId 为 0 的 MessageQueue，因为一个 Topic 下只有 1 个 MessageQueue：

• 这个 Topic 下的所有 Message 就是全局有序的，ta 们会按照先来后到的顺序被消费

如果本地事务执行成功进行 Commit，则将 RMQ_SYS_TRANS_HALF_TOPIC 队列中的消息投递到真实的 Topic 中，供后续流程执行

• 并删除这条 Half Message，但删除也是假删除，只是给 Message 打上一个删除的 tag

如果本地事务执行失败进行 rollback，则直接删除这条 Half Message，但删除也是假删除

如果本地事务吃吃没有返回结果（默认时间 6s），则会触发事务回查机制

• 执行回查之前需要校验检查次数是否达到了最大值（需要手动设置，没有默认值）
• 或者是当前 Half Message 存在是否超过了 Message 保存的上限，即 3 天
• 如果满足上面条件中的一种 Half Message 会被放进 TRANS_CHECK_MAX_TOPIC Topic 中
• 一旦判定为需要执行事务回查逻辑，那么当前这条 Half Message 就算已经被消费了
• 在没有达到最大的校验次数之前，都还需要将其投递到事务队列当中，以便下次重试时再次执行 Check 逻辑
• 如果回查成功，则删除投递的 Half Message

消息重试

重试时间

消息消费失败后，并不会立即重试，而是一个递增的时间间隔来进行重试的，重试次数默认为 16 次

只比延迟消息的时间间隔等级少了前两个，延迟消息总共有 18 个等级，而消息重试使用了延迟消息的第 3-18 等级

10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

基本原理

重试的 Message，RocketMQ 的做法并不是将其投递回原来的 Topic，而是重试队列

每个 ConsumerGroup 都有自己的重试队列：

• 其名称是由特定的前缀拼接上 ConsumerGroup 所组成，默认 %RETR% + 消费者组名称
• 所有在 Consumer 启动时，就会同时消费其 ConsumerGroup 对应的重试队列和普通队列

消费失败的 Message，Consumer 会将其投回 Broker：

• 相当于这条 Message 已经被消费掉了，之后重试的只是内容相同，但实际不是同一个的 Message
• 然后会校验重试的次数，如果达到 16 次，则会进入死信队列，组成为 %DLQ% + 消费者组名称
• 未达到最大重试次数，则会根据重试间隔时间等级将其投递到延迟队列 SCHEDULE_TOPIC_XXXX 中
• 然后等到了延迟等级对应的时间后，在投递到 ConsumerGroup 所对应的重试队列当中，供后续消费

消息存储

整体架构

RocketMQ 的混合性存储结构（多个 Topic 的消息实体内容都存储于一个 CommitLog中）

针对 Producer 和 Consumer 分别采用了数据和索引部分相分离的存储结构

Producer 发送消息至 Broker 端，然后 Broker 端使用同步或者异步的方式对消息刷盘持久化，保存至 CommitLog 中

核心步骤：

1. 首先，生产者根据 topic 发送消息，消息存储在 commitLog中，1 G一个文件，当文件满了，写入下一个文件
2. 其次，ReputMessageService 重写消息服务执行 2 个分发操作：
   • 创建 ConsumerQueue 逻辑消费队列：
     • 参数：commitLogOffset 物理偏移量、msgSize 消息长度、tagsCode tag 哈希
   • 创建 IndexFile 索引文件：
     • 以创建时的时间戳命名
3. 最后，消费者根据 topic、tag 拉取消息消费，根据 key 查询消息

重要文件

commitLog 消息日志：

• 消息主体以及元数据的存储主体，存储 Producer 端写入的消息主体内容

consumequeue 逻辑消费队列：

• 存储了 commitLog 的起始物理 offset，目的是提高消费的性能

indexFile 索引文件：

• 提供了一种可以通过 key 或者时间区间来查询消息的方法

consumequeue 文件：

consumequeue 文件采取定长设计，每一个条目共 20 个字节，分别为：

• 8 字节的 commmitLog 物理偏移量
• 4 字节 的消息长度
• 8 字节 tag hashcode

单个文件由 30w 个条目组成，可以像数组一样随机访问每一个条目，每个 ConsumeQueue 文件大小约 5.72M

• 默认一个 topic 对应 4 个 queueId，即 4 个 messageQueue

每个 messageQueue 文件夹下有多个 consumeQueue，所以：messageQueue 1 ：N consumeQueue

通信机制

通信架构图

基本通讯流程如下：

• Broker 启动后需要完成一次将自己注册至 NameServer 的操作，随后每隔 30s 时间定时向 NameServer 上报 Topic 路由信息
• 消息生成者 Producer 作为客户端发送消息时，需要根据消息的 Topic 从本地缓存的 TopicPublishInfoTable 获取路由信息
  • 如果没有则更新路由信息会从 NameServer 上重新拉取，同时 Producer 会默认每隔 30s 向 NameServer 拉取一次路由信息
• 消息生产者 Producer 根据获取的路由信息选择一个队列（MessageQueue） 进行消息发送
  • Broker 作为消息的接收者接收消息并落盘存储
• 消息消费者 Consumer 根据获取的路由信息，并再完成客户端的负载均衡后，选择其中的某一个或者某几个消息队列来拉取消息并进行消费

为了实现客户端与服务器之间高效的数据请求与接收：

• RocketMQ-Remoting 包自定义了通信协议并在 Netty 的基础之上扩展了通信模块