“怎么还没收到短信？”RocketMQ 堆了 300 万条消息，我从 Diff Total 一路追到一个死循环重试

周五下午 5 点 40，客服群里冒出一条消息：“有用户反馈，验证码半小时没收到，申诉的电话已经打到老板那里了。”我们的短信是走 MQ 异步发的：业务系统往 RocketMQ 丢一条消息，短信服务消费后调用运营商网关发送。链路里任何一环慢了，用户的感受都只有一个——“怎么还没收到？”

我打开 RocketMQ 控制台，看到 sms-send 这个 Topic 的消费组，堆积数字一栏写着：3,127,406。我盯着这串数字沉默了三秒，然后默默把咖啡放下——今天的周末，大概没了。

一、现象：堆积，而且还在涨

先说一个基本概念，后面所有判断都建立在它上面：

• Broker Offset：某个队列（queue）当前写到的位置（生产进度）；
• Consumer Offset：消费组在这个队列已经确认消费到的位置（消费进度）；
• Diff = Broker Offset − Consumer Offset：还没被消费的消息数，也就是我们常说的“堆积量”；所有队列的 Diff 加起来就是 Diff Total。

堆积本身不是故障，业务高峰时短暂堆积再消化完很正常。判断是否有问题，要看趋势：Diff Total 是在涨、持平还是在降。涨，说明消费速度 < 生产速度，或者根本没在消费。

所以排查 MQ 堆积，其实就是回答一连串问题：是生产变多了，还是消费变慢了，还是消费者干脆掉线了？慢在哪一环？

二、排查路线图

三、动手：mqadmin 三板斧

控制台好看，但值班时我更信命令行：输出稳定、可复制进工单、可以拿来做前后对比。RocketMQ 自带的运维工具是 bin/mqadmin，先设个环境变量省得每次写 NameServer 地址：

$ export NAMESRV=10.0.0.11:9876
$ cd /opt/rocketmq/bin
$ sh mqadmin consumerProgress -n $NAMESRV -g sms-send-consumer-group
#Topic          #Broker Name  #QID  #Broker Offset  #Consumer Offset  #Client IP      #Diff    #LastTime
sms-send        broker-a      0     18823411        17712930          10.0.3.21       1110481  2024-10-11 17:12:03
sms-send        broker-a      1     18820977        17709112          10.0.3.21       1111865  2024-10-11 17:12:03
sms-send        broker-a      2     18819322        17711876          10.0.3.22       1107446  2024-10-11 17:12:04
sms-send        broker-a      3     18821050        18821050          10.0.3.22       0        2024-10-11 17:41:55
...
Consume TPS: 21.35
Diff Total: 3127406

各列含义（不同版本列略有差异，比如 4.9+ 会多出 Inflight 等列，以 mqadmin consumerProgress -h 和实际输出为准）：

• #Broker Offset / #Consumer Offset / #Diff：这个队列上的写入位置、消费位置、堆积数；
• #Client IP：当前负责消费这个队列的消费者实例（负载均衡结果）；
• #LastTime：这个队列最近一次被消费的消息的存储时间——它是判断“卡多久了”的好线索，如果它远远落后于当前时间，说明消费落后了那么久；
• Consume TPS：这个消费组当前的消费速度（条/秒）；
• Diff Total：所有队列 Diff 之和，也就是堆积总量。

这份输出里有好几个值得注意的点：

1. Consume TPS 只有 21。平时高峰我们的短信量是每秒几百条，21 明显不对；
2. 三个队列各堆了 110 万，QID 3 的 Diff 是 0：说明这个队列被消费得“很顺”（或者没有生产数据），并不是所有队列都堵；
3. Client IP 只有 10.0.3.21 和 10.0.3.22 两个——我们明明部署了 4 个消费者实例。

第 3 点很可疑，我们进入第二板斧：看消费者连接。

第 2 步：consumerConnection——消费者是不是还活着、订阅对不对

$ sh mqadmin consumerConnection -n $NAMESRV -g sms-send-consumer-group
#Consumer Connection List
#ClientId                    #ClientAddr         #Language  #Version
10.0.3.21@sms-svc-7c9d5      10.0.3.21:41822     JAVA       V4_9_4
10.0.3.22@sms-svc-5f8b2      10.0.3.22:38160     JAVA       V4_9_4

Below is subscription:
#Topic          #SubExpression
sms-send        *
%RETRY%sms-send-consumer-group  *

ConsumeType: CONSUME_PASSIVELY
MessageModel: CLUSTERING
ConsumeFromWhere: CONSUME_FROM_LAST_OFFSET

输出说明了几件事：

• 只有 2 个消费者连接（另外 2 个实例的 pod 被驱逐后没有正常重启，掉线了）——它们对应的队列由剩下的 2 个实例通过重平衡（rebalance）接管，消费能力直接减半；
• 订阅里除了业务 Topic，还有一个 %RETRY%消费组名：这是 RocketMQ 自动为消费组订阅的重试队列。消费失败的消息会进入这里，稍后再投递。后面会看到，它是这次事故的关键；
• MessageModel: CLUSTERING 是集群消费，一条消息只会被消费组里一个实例消费；CONSUME_PASSIVELY 表示 Push 模式（实际是客户端长轮询拉取）。

如果这里显示空列表（或报“the consumer group not online”），那就是消费者全部掉线，Consume TPS 会是 0，Diff 一路上涨——此时要去查实例为什么没起来，以及 订阅关系是否一致（同一消费组的不同实例订阅了不同 Topic/Tag，会互相覆盖，造成部分消息永远没人消费）。

第 3 步：topicStatus——看生产端和存储进度

$ sh mqadmin topicStatus -n $NAMESRV -t sms-send
#Broker Name   #QID  #Min Offset  #Max Offset  #Last Updated
broker-a       0     16810022     18823980     2024-10-11 17:42:10
broker-a       1     16805411     18821544     2024-10-11 17:42:09
broker-a       2     16807290     18819877     2024-10-11 17:42:10
broker-a       3     16809933     18821601     2024-10-11 17:42:10

这里能看到每个队列写入位置（Max Offset）的增长以及最后更新时间——说明生产端仍在正常写入。隔一分钟再执行一次，比较 Max Offset 的增量，就能算出生产速度：我们这次每个队列每分钟增长约 500 左右，4 个队列合起来大约每秒 33 条。消费 TPS 只有 21，21 < 33，堆积自然持续增长。

另一个有用的工具是查某条具体消息的状态，业务方常拿一个 msgId 或 key 来问“这条消息到哪了”：

$ sh mqadmin queryMsgByKey -n $NAMESRV -t sms-send -k SMS20241011000123
$ sh mqadmin queryMsgById  -n $NAMESRV -i 0A0000030000...

producer 发送时设置 message.setKeys(业务单号) 是官方最佳实践之一，能让排查时按业务单号找到消息，别偷懒。

四、继续钻：消费为什么这么慢

现在的判断是：消费者少了一半 + 消费 TPS 只有平时的十分之一。少一半解释不了“十分之一”。剩下的实例上到底发生了什么？登上其中一个消费者的机器，抓线程：

$ jps -l | grep sms
8231 com.demo.sms.SmsApplication
$ jstack 8231 | grep -A 12 'ConsumeMessageThread' | head -50
"ConsumeMessageThread_sms-send-consumer-group_7" #93 prio=5 os_prio=0 nid=0x2a41 runnable
   java.lang.Thread.State: RUNNABLE
	at java.net.SocketInputStream.socketRead0(Native Method)
	at java.net.SocketInputStream.read(SocketInputStream.java:171)
	...
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:108)
	at com.demo.sms.gateway.OperatorClient.send(OperatorClient.java:64)
	at com.demo.sms.consumer.SmsConsumer.consumeMessage(SmsConsumer.java:41)
	at org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(...)
"ConsumeMessageThread_sms-send-consumer-group_8" #94 ... runnable
   ... (同样卡在 socketRead0 / OperatorClient.send)

20 个消费线程几乎全部卡在 socketRead0，调用栈指向 OperatorClient.send——运营商短信网关。也就是说线程没死，是在等下游响应：那天网关的响应从平时 50ms 变成了 2~3 秒，而我们的 HttpClient 读超时配的是 30 秒。粗算一下：20 个线程 ÷ 平均 1 秒 / 条 ≈ 20 条/秒——和 Consume TPS 21 严丝合缝。

这里顺手记一个通用公式：消费 TPS ≈ 消费线程数 ÷ 单条消费平均耗时（秒）。TPS 上不去，要么线程少，要么单条太慢，别的都是玄学。

再看重试队列：越积越多的“失败消息”

接着，我在 consumerProgress 的输出里往下翻，发现了这一段：

$ sh mqadmin consumerProgress -n $NAMESRV -g sms-send-consumer-group | grep RETRY
%RETRY%sms-send-consumer-group  broker-a  0  2208113  2118902  10.0.3.21  89211  2024-10-11 17:40:55
Diff Total: 3127406

重试队列本身也堆了 8.9 万。它是怎么来的？看消费代码：

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) {
    for (MessageExt msg : msgs) {
        try {
            operatorClient.send(parse(msg));         // 调运营商网关，可能超时
        } catch (Exception e) {
            log.error("send sms failed", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;   // 任何异常都重试
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

逻辑是“失败就让 RocketMQ 稍后重投”，看起来很稳。但结合“下游变慢、大量超时”这个场景，问题就来了：

1. 网关超时 → 抛异常 → 返回 RECONSUME_LATER；
2. 这条消息被送进 %RETRY% 队列，按退避策略在一段时间后再次投递：官方文档里，PushConsumer 无序消息的重试间隔是递增的，从 10 秒开始，逐步拉长（30 秒、1 分钟、2 分钟……），最多 16 次左右，最长间隔 2 小时；
3. 重试消息回来时，又要占用消费线程去调那个已经很慢的网关——“失败消息”和“新消息”抢同一批线程，而这批线程已经被慢下游拖住了；
4. 一批一批消息在重试队列里反复兜圈，占用消费能力，直到网关恢复或消息重试次数耗尽进入死信队列。

更糟的是，我们的 catch (Exception e) 把所有异常（包括“手机号格式非法”这种永远不可能成功的）统统丢去重试。这类毒消息会白白重试 16 次，一条消息占用一段漫长的“无效消费”。当时我在日志里 grep 出这样的内容：

$ grep 'send sms failed' /data/logs/sms/app.log | awk '{print $NF}' | sort | uniq -c | sort -rn | head -3
  48213 java.net.SocketTimeoutException: Read timed out
   2014 com.demo.sms.gateway.InvalidPhoneException: invalid mobile number
     56 java.lang.NullPointerException

四万八千次读超时是“下游慢”；两千多次InvalidPhoneException 是“毒消息”，永远不会成功却反复重试；还有几十个空指针，是解析逻辑的 bug。

五、根因

把因果链串起来：

1. 触发：运营商网关变慢（50ms → 2~3s）；
2. 放大 1：读超时配 30 秒，消费线程被长期占用，Consume TPS 从几百掉到 21；
3. 放大 2：两个消费者实例掉线，可用消费线程再减半；
4. 放大 3：所有异常都 RECONSUME_LATER，失败/毒消息挤占线程，重试队列自己也在堆积；
5. 结果：生产速度 33 条/秒 > 消费速度 21 条/秒，Diff Total 缓慢却坚定地涨到 300 万。

顺带回答一个高频问题：“队列多、消费者少 vs 消费者多、队列少”，谁是并行度上限？在 RocketMQ 集群消费模式下，一个队列同一时刻只会被消费组内的一个消费者实例拉取消费（重平衡的分配单位是“队列”）。所以实例数超过队列数，多出来的实例是空转的。Topic 只有 4 个队列（我们那个 Topic 就只有 4 个队列，历史遗留），那么 10 个实例也只有 4 个有活干；想提升整体并行度，得先加队列（updateTopic -r -w），再加实例；而单个实例内的并行度则由消费线程数决定。

六、止血与修复

止血（按代价从低到高）

1. 先把慢下游的影响隔离：把 HttpClient 读超时从 30 秒改成 3 秒，连接超时 1 秒；网关慢时快速失败，线程不被长期占住。通过配置中心热更新，不用发版；
2. 恢复掉线的实例：把被驱逐的两个 pod 拉起来，4 个实例齐活，重平衡后每个实例负责 1 个队列；
3. 提高实例内并行度：consumeThreadMin/consumeThreadMax 从 20 调到 64（消费逻辑是 IO 密集型，线程数可以适当放大，但要评估下游能否承受）；
4. 启用批量消费：consumeMessageBatchMaxSize 默认 1，改为 10~32，让一次回调处理多条消息，对“批量入库”“批量调用”类场景效果明显（注意：一批里有一条失败，会整批重试，需要业务做好幂等）；
5. 历史消息是否还有价值？验证码短信超过 5 分钟就已经失效了，堆了 30 分钟的验证码继续发只会骚扰用户。此时最好的策略是跳过积压：把消费位点直接重置到当前时间。

# 把消费位点重置到“现在”，跳过所有积压（不可逆！先确认业务方同意，并保存好当前 consumerProgress 输出）
$ sh mqadmin resetOffsetByTime -n $NAMESRV -g sms-send-consumer-group -t sms-send -s now
rollback consumer offset by specified group[sms-send-consumer-group], topic[sms-send], force[true], timestamp(string)[now], timestamp(long)[1728639845123]
#brokerName     #queueId    #offset
broker-a        0           18829977
broker-a        1           18827413
broker-a        2           18826120
broker-a        3           18827701

resetOffsetByTime 的 -s 支持 now、毫秒时间戳或 yyyy-MM-dd#HH:mm:ss:SSS 格式，只是移动消费位点，不删除消息（消息仍按保留策略——默认 72 小时——保存在 CommitLog，需要时可以把位点再拨回去补发）。执行前请确认消费组在线情况（Push 消费者在线时，Broker 会通知客户端同步位点；不在线时也可以重置，等下次上线生效）。

如果历史消息必须消费（比如账单、订单通知），就不能跳，只能扩容硬追。这时按公式估算追平时间：积压 ÷ (消费 TPS − 生产 TPS)。300 万 ÷ (400 − 33) ≈ 8170 秒 ≈ 2.3 小时。想缩短，只有加线程、加队列、加实例、优化单条耗时。

根治 1：区分“可重试”与“不可重试”，失败有上限，进死信

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) {
    for (MessageExt msg : msgs) {
        try {
            operatorClient.send(parse(msg));
        } catch (InvalidPhoneException | IllegalArgumentException e) {
            // 业务上永远不会成功：记录后直接确认，别去重试
            log.warn("drop invalid sms, key={}, reason={}", msg.getKeys(), e.getMessage());
        } catch (SocketTimeoutException e) {
            // 可恢复：让 RocketMQ 稍后重投
            if (msg.getReconsumeTimes() >= 3) {
                // 重试 3 次仍失败：落库，走补偿任务，不再占用重试队列
                failRepo.save(msg.getMsgId(), msg.getKeys(), e.getMessage());
            } else {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

关于官方对重试与死信的规则，需要记住：消费失败的消息自动重试，达到最大次数后进入死信队列，死信队列的 Topic 名称是 %DLQ%消费组名，其中的消息不会再被消费者自动消费，需要人工排查处理（如通过控制台重发或写脚本消费死信 Topic）。重试次数也可以在消费组层面配置，并不是所有版本都固定 16 次。查看死信队列：

$ sh mqadmin topicList -n $NAMESRV | grep DLQ
%DLQ%sms-send-consumer-group
$ sh mqadmin topicStatus -n $NAMESRV -t %DLQ%sms-send-consumer-group
#Broker Name   #QID  #Min Offset  #Max Offset  #Last Updated
broker-a       0     0            2014         2024-10-11 17:31:22

2014 条，正好对应前面 grep 出来的“非法手机号”数量。死信队列存在的意义就是把有毒的消息隔离出来，别让它们反复消耗正常消费的资源。（死信消息默认也是保留 3 天，别等它被清了才发现，要给死信 Topic 加告警。）

根治 2：下游调用加超时、隔离与熔断

RequestConfig cfg = RequestConfig.custom()
        .setConnectTimeout(1000)         // 连接 1s
        .setSocketTimeout(3000)          // 读 3s（原来是 30s）
        .setConnectionRequestTimeout(500)
        .build();
// 配合 Resilience4j / Sentinel：网关错误率超过阈值时熔断，快速失败 → RECONSUME_LATER，
// 避免 20 个线程全被同一个慢下游占满

根治 3：容量与监控

• 队列数留余量：Topic 队列数决定并行度上限，预估峰值所需实例数后，再乘以 2 作为队列数（如 16 个队列），扩容时才有空间；
• 监控指标：消费组堆积量（Diff Total）、消费延迟时间（当前时间 − 最新消费消息的存储时间，比堆积条数更贴近业务感受）、Consume TPS、重试队列与死信队列的增量；
• 告警要看趋势：“堆积 > 10 万且 5 分钟内持续增长”比“堆积 > 10 万”更有价值；
• 消息设置 Keys：便于用 queryMsgByKey 追踪某条业务消息。

七、验证

$ sh mqadmin consumerProgress -n $NAMESRV -g sms-send-consumer-group | tail -3
sms-send        broker-a      3     18827950        18827948        10.0.3.24       2        2024-10-11 18:21:40
Consume TPS: 36.10
Diff Total: 7

调整超时、恢复实例、重置位点之后，Diff Total 从 300 万降到个位数，消费 TPS 与生产 TPS 持平。整个过程 40 分钟，用户端最晚收到新验证码的延迟不到 5 秒。

八、复盘

一句话备忘：堆积三问——Consume TPS 有没有？（consumerProgress）消费者在不在？（consumerConnection）线程卡在哪？（jstack）。扩容不超过队列数，失败重试要有上限进死信，过期消息直接 resetOffsetByTime -s now 跳过。

那天晚上我在群里发了句“已恢复”，客服回了个“辛苦了”，产品回了个“下次能提前发现吗”。我看着监控里那条终于回到 0 的曲线，心想：可以，但得先给 MQ 装个“会喊疼”的告警——堆到 300 万才有人喊，说明我们的消息队列太能忍了。

参考资料

• Apache RocketMQ 4.x：Basic Best Practices——消费幂等、处理慢消费（增加实例/线程、批量消费、跳过非重要消息）、RECONSUME_LATER、消息设置 Keys、SendResult 状态。
• Apache RocketMQ：Consumption Retry——最大重试次数、PushConsumer 无序消息的递增重试间隔、超过最大次数进入死信队列的状态机。
• 博客园：rocket mq 查看消费进度、消息堆积、清除堆积数据命令——consumerProgress 的 Broker Offset / Consumer Offset / Diff 含义，以及 resetOffsetByTime 重置位点与 ConsumeFromWhere 说明。
• 腾讯云开发者社区：常用 RocketMQ 命令——consumerConnection、consumerProgress、topicStatus、queryMsgByKey 等 mqadmin 命令的用法。