RocketMQ 5.X PopAck 源码拆解

RocketMQ 已经开启 5.X 时代，4.X 已成为 LTS 版本。

各大云厂商也推出支持 RocketMQ 5.X 版本的产品，在介绍 Pop，Ack 之前需要先了解 RocketMQ 5.X 的架构。

如上图，在 RocketMQ 5.X 的架构中，新增组件如下：

• 02 Controller：控制器，帮助 Broker 做主从切换。
  

• 04 Proxy：RocketMQ 的代理服务，支持 gRPC 协议客户端、Remoting 客户端收发消息。
  

• 05 gRPC Client：RocketMQ 5.X 的新客户端，使用 gRpc 协议访问RocketMQ Proxy。

注意：社区很多朋友咨询这个客户端是否可以访问 4.X 集群，答案是不支持。

5.X 其实有2个客户端，优先推荐使用 gRPC 客户端：

5.X gRPC客户端源码: https://github.com/apache/rocketmq-clients

5.X Remoting客户端源码：

https://github.com/apache/rocketmq/tree/develop/client

5.X gRPC 客户端使用 gRPC 协议访问 Proxy，5.X Remoting 客户端可以使用 Remoting 协议访问 Namesrv 和 Proxy。

其余组件是 RocketMQ 4.X 的原组件。想要了解更多 5.X 和 4.X 的差别，请看：RocketMQ 5.0 vs 4.9.X 图解架构对比。

Pop主要解决 Push、Pull 消费者常见的4种问题：

• 消费者卡住问题

下图是 Push 消费者订阅 Consumer Queue 的情况，如果 Push Client 2客户端由于 GC 等原因执行特别慢，此时 Broker 1-1 中 Queue 1，Broker2-1 中的Queue1 将出现堆积。

• 负载均衡慢的问题。

如果出现 Push Consumer 卡住，或者 GC 导致消费慢，此时我们一般通过重启消费者程序来临时解决。

消费者重启后 Reblance，消费者数量越多，Reblance 花费的时间越长，在 Reblance 时消费者无法消费消息。

Pop 消费时，一个消费者的上线、下线不会触发 Reblance，故而不存在负载均衡慢的问题。

• 有状态变无状态。

Pop 是无状态消费者。在云原生大环境下，无状态服务在扩/缩容时更方便、快捷。

• 消费实例数上限问题。

Push 消费者的消费者个数最大不能超过 Consumer Queue 的个数，在 Pop 时不再有这个限制，但是也有自身限制。

Broker 在 Pop 时通过 Lock Consumer Queue 实现 Pop 消息，多个 Pop 消费者客户端进行锁竞争的时间和 Pop 消费者的个数成此起彼伏的形态，故而 Pop 消费者个数不能无限扩张。

如何锁 Consumer Queue，接下来讲解。

在 Broker 中，Pop 的实现代码从

PopMessageProcessor.processRequest(ChannelHandlerContext, RemotingCommand) 方法开始的。

笔者把 Pop message 主要分为5个流程，这里忽略数据校验、Pop 前参数准备。

这里大家直接看代码 PopMessageProcessor.processRequest()中调用了 PopMsgFromQueue()方法，此方法为 Pop 消息的实现关键，笔者将其分解为5步：

1、锁 Consumer Queue

String lockKey =            topic + // topic名字            PopAckConstants.SPLIT +  // 分隔符            requestHeader.getConsumerGroup() + // 消费者组            PopAckConstants.SPLIT +  // 分隔符            queueId; // consumer queue id

可以得知：一个 Consumer Queue 同时只会被同一个消费者组中的某一个1个消费者实例锁住。

如果同一个消费者组中同时有2个消费者实例来 Pop 消息，只有一个会锁成功。

这个锁的实现代码如下：

关于这把锁有2点需要注意：

• 这是一把 TimedLock。 锁是有超时时间的，超过锁的时间自动释放。通过图中标注1可以看出, 当前锁服务是一个 ServiceThread，在 RocketMQ 中这个代表这个服务是一个后台线程，会自动执行检查。

• 高效锁。锁是通过 ConcurrentHashMap 的线程安全实现的，在一般的 Java面试中相信大家也经常被问到。

2、计算 Pop Offset

在 Pop 消息流程中有2处地方计算了 Pop Offset。

• 第一处。在加锁 QueueXX 前计算 Pop Offset。在加锁失败时，会根据 Pop Offset 估算在这个 Consumer Queue 中还有多少个消息没有被消费。

也就是返回字段 ：RestNum。

• 第二处。在加锁 QueueXX 成功后计算 Pop Offset。

为什么需要重新计算？从第一次执行计算 Pop Offset 到加锁成功后这段时间可能有其他人更新了消费位点，导致第一次计算的 Pop Offset 不准确，在加锁 QueueXX 成功后，QueueXX 只会被当前客户端 Pop 消息，此时重新计算 Pop Offset 的值是准确值，根据这个再去 Store 中读取消息。

3、读取消息

读取消息调用this.brokerController.getMessageStore().getMessageAsync()方法读取，下次有机会再细聊。

4、添加 Check point

Check Point 消息（简称 CK 消息），其中记录了每次 Pop 的消息信息。

在读取完消息后，将生成一个 Check point 消息（简称 CK 消息）。CK 消息将写入一个 Buffer 中。调用代码如下：

写入 Buffer 后，这些消息将进入不可见时间，也就是同一个消费者组的其他消费者实例无法再读取到，为什么呢？

在写入 Buffer 后，其他消费者实例计算 Pop Offset 时，会把 Buffer 中已经 Pop 的消息计算在内，所以就不会读取到消息。

如果不可见时间到了，用户也没有 Ack，这些被 Pop 的消息会被 Revive 服务再次恢复到用户的 Topic 中被用户消费

5、释放 Consumer queue 锁

这个没啥说的，直接看加锁的那个图。

• Pop Offset

Pop Offset 是每个消费者实例在 Pop 消息的时候会计算的，被 Pop 的 Queue 中可以被消费的消息的起始位点。

RocketMQ 会用这个 Pop Offset 去存储中读取消息。

读取消息的流程和 4.X 差不多，这里不再赘述。

下图展示了 Pop Offset 是如何计算的：

• Check point

public class PopCheckPoint implements Comparable<PopCheckPoint> {// 本次pop消息d的起始consumer queue offsetprivate long startOffset;     // 本次pop时的时间戳，单位毫秒private long popTime;// 本次pop消息d的不可见时间，单位毫秒    // 一般来自pop客户端请求的request headerprivate long invisibleTime;    // 特别重要    // 记录本次pop消息的ack情况private int bitMap;// 本次pop消息d的条数private byte num;// 本次pop的consumer queue idprivate int queueId;    // 本次pop 的topicprivate String topic;    // 本次pop 的消费者组private String cid;// 特别重要    // revieve topic的位点，后面详细讲解private long reviveOffset;// 特别重要    // 本次拉取消息d的每个消息的queue offset 减去 pop offset    // 的差值private List<Integer> queueOffsetDiff;// 本次pop 消息所在d的brokerprivate String brokerName;}

• BitMap

这个字段是一个 int 类型，1个 Int 是由32个 Bit 表示，每个 Bit 其实就是0，1，RocketMQ 利用 BitMap 标记本次 Pop 的消息哪些被 Ack（标记为1），哪些未 Ack（标记为0）。具体过程详见后面讲解 Ack 的过程。

• ReviveOffset

Revive 英文翻译是恢复的含义，那些不可见消息基础信息（非消息 Body）会保存到 Revive Topic 中，到时间后会被 Revive 服务恢复到用户的原始 Topic 中让用户再次消费。ReviveOffset 就是这个 Revive Topic 的 Consumer Queue 位点。

• QueueOffsetDiff

• ReceiptHandle

这个值叫一条消息的句柄，每个消息一条，Ack 的时候会给到 Broker，Broker 通过解析判断 Ack 的哪次 Pop 的哪条消息， 格式如下：

通过上图我们可以知道，所谓的句柄其实是消息的一堆属性拼接起来的一个字符串。

这个字符串实际长这个样子：

• StartOffsetInfo

这个值是一次 Pop 一个值， 记录了 Pop 的起始位点信息， 实际格式如下：

• MsgOffsetInfo

• OrderCountInfo

这个值是一次 Pop 一个值， 记录了 Pop 顺序消息的每个消息 Reconsume Times，格式如下：

Proxy 中使用的代码如下：

通过以上核心数据结构，我们可以看出来：Broker 针对 Pop 输出了很多数据结构给 Proxy 使用。

这里笔者也有一个疑问：这些数据结构加大了 Proxy 与 Broker 的耦合逻辑，这使得 Proxy 做纯粹的无状态变得困难。

是否可以只做到接口耦合，不用做逻辑耦合？

Ack 是针对 Pop 的， 一次 Pop 可以 Pop 出多条消息， 但是 Ack 的需要解决以下几种情况：

• 用户每次 Ack 一条消息。

• 用户每次 Ack 一批消息。

• 用户出错了， 永远没有 Ack。

上面3种情况的结果有4种：

• 用户 Ack 完成了这次 Pop 的全部消息。

• 用户 Ack 完成了部分消息， 并且 Ack 的位点有空洞。

• 用户 Ack 完成了部分消息， 并且 Ack 的位点无空洞。

• 用户没有 Ack 任何一条消息。

基于上面的几种场景， RocketMQ Ack 是如何实现的呢？

笔者总结了 Ack 的流程：

在 Broker 中， Ack 的入口是 AckMessageProcessor.processRequest() 方法，其中虚线是异步的流程，实线是同步流程。笔者将其分为以下5步。

第一步：Proxy 提交 Ack 请求

用户提交 Ack 请求，Ack 请求被 Broker 的AckMessageProcessor.processRequest(Channel, RemotingCommand, boolean)方法处理，并解析 AckMessageRequestHeader。

AckMessageRequestHeader 中包含 Pop ck 信息， 这里逻辑上区分单个消息Ack 还是批量消息 Ack：

标记1: 单个消息 Ack。

标记2: 批量消息 Ack。

标记3: AppendAck() 方法是 Ack 核心逻辑，后面的全部逻辑都在这个方法中实现。

标记4: 批量执行 AppendAck() 方法。

可以看到标记4处理非原子操作是一种风险，批量提交结果未知，以最终结果一致为准。

第二步：内存标记消费进度

经过第一步后，我们知道核心逻辑在 AppendAck() 中： RocketMQ 将 Ack Request Header 解析为 AckMsg，并且调用 PopBufferMergeService.addAk() 将 AckMsg 写入 PopBufferMergeService 的缓存中。

PopBufferMergeService 顾名思义，是一个在内存中提供合并的服务。

合并什么呢， 合并 Ack和 CK 消息，也就是用 Ack 的 Consumer Queue Offset 去标记 CK 中的 BitMap。

其实就是标记一个 CK 中的哪些消息被 Ack 了，也就是标记了消费进度。

• 假设拉取了4个消息,组成一个数组，每个“消息的下标”分别为:0,1,2,3
• 4个消息是否消费的标记由4个“二进制标记”组成一个数组
• 二进制标记数组，可以转化为“1个10进制数int”保存ck对象中

图示如下：

• 全部的消息都 Ack。
• 用户在允许的时间内没有 Ack 完成全部消息。
• 用户 Ack 的时候， Check Point 消息已经不存在了。

• 内存不能保存全部的 CK。Pop Check Point 信息会保存到内存中， 这里不可能保存全部的 Pop Check Point， Broker 提供配置 PopCkMaxBufferSize 内存最大可以保存的 Pop Check Point 数，默认20w。

• 允许时间内没有 Ack 的的 CK 需要丢弃，这个 CK 对应的全部消息全部对用户再次可见。

第四步：异步标记消费进度

经过上一步，我们知道有一些 Check Point 信息和 Ack 信息会被持久化到 Revieve Topic。

PopBufferMergeService 服务是一个后台服务， 会消费 Revieve Topic 中的 Ack、CK 信息，然后做异步匹配， 来标记 CK 信息中的用户消息哪些被 Ack 了。

这里细节特别多， 建议大家 Debug 查看，这里如果需要细讲大家留言我们再出一期。

经过 Scan 后，可以知道哪些 CK 中的用户消息被全部 Ack了， 就会提交消费位点到 Consumer Queue Offset Manager：

如果经过这一步，还是有 CK 没有完全被 Ack 呢？请看下一步。

第五步：可见时间过后，消息恢复消费

如果经过上一步还有 CK 没有被 Ack 完全匹配，此时这些 CK 对应的用户消息将被重新可见，用户可以重新 Pop。

这个过程是在 PopReviveService 服务中实现的， 这也是一个后台服务， 会定时检查哪些 CK 没有被完全 Ack， 然后根据 CK 将这个 CK 包含的全部消息重新恢复到重试 Topic 中。

• 1. 同一个 Pop CK，多次重复 Ack 会出现什么情况， Broker 是如何处理的？
• 2. 如果 Pop 没有读取到消息需要写 CK 信息吗， 为什么？
• 3. 下期准备讲 Proxy 或者基于时间轮的任意定时消息，想看什么请留言。

往期

推荐

《Apache Pulsar 在腾讯云上的最佳实践》

《KMS 在腾讯云的微服务实践助力其降本50%》

《Apache Pulsar 技术系列 - PulsarClient 实现解析》

《小鹅通基于 TSE 云原生 API 网关的落地实践》

《腾讯云微服务平台 TSF 异地多活单元化能力重磅升级》

《微服务高可用容灾架构设计》

扫描下方二维码关注本公众号，

了解更多微服务、消息队列的相关信息！

解锁超多鹅厂周边！

戳原文，查看更多消息队列 RocketMQ 版的信息！

点个在看你最好看