已采纳
回答正文
回复
讨论正文
该问题的正文与回答。
【背景】
我们用 RocketMQ 4.9.7(3 broker 集群)承载支付清结算的状态机流转:topic settle-state,16 个 queue,消费组 4 实例,使用 MessageListenerOrderly 保证同一笔结算单内严格顺序。下游是 MySQL 8.0(一主两从)+ 一次 Redis 写 + 一次外部风控 HTTP 调用(P99 约 350ms,偶发 8s 以上超时)。
【现象】
过去两周出现 3 次「顺序消费整体停摆」:表现为该消费组 lag 从 0~30 一路涨到 50 万+,且所有 16 个 queue 同时出现 lag(不是某一个 queue 慢),消费侧线程全部卡在 MessageListenerOrderly#consumeMessage 内部的 synchronized(看 jstack 能确认),业务上结算单状态停滞 40~90 分钟,最终靠重启消费实例恢复。
【已尝试的排查与结果】
MessageListenerOrderly 的锁上,但锁对象是 per-queue 的 ConcurrentHashMap,没法从堆栈直接读出 queueId 和 msgId;mqadmin:用 consumerProgress 能看到每个 queue 的 consumerOffset / brokerOffset,但卡住时所有 queue 都「落后」,无法区分「谁是根因 queue、谁是被动阻塞」;【期望得到的答案形态】
queryMsgById / consumerProgress 里,在不停机的前提下锁定「根因 queueId + 卡住的那条 msgId」?MessageListenerOrderly 里加「单条消息处理超时熔断 + 告警」的轻量做法,不破坏顺序语义?(比如 Future.get 超时、或把外部调用挪到消息外?)