★ FEATURED ARTICLE 消息队列堆积不是一个会不会发生的问题而是什么时候发生、发生之后多久能恢复的问题。只要生产速率在某个时间窗口超过消费速率堆积就会开始累积大促流量高峰、下游数据库抖动、消费端发布重启、某条消息反复重试卡住分区随便一个都会把 Consumer Lag 拉起来。这篇文章不讲消息队列的基础概念重点聊工程上真正用得上的部分怎么判断堆积到了什么程度、先做哪一步能最快把水位压下去、消费端有哪些可以直接改的并发与批量参数、Broker 侧有哪些容易忽略的坑以及堆积恢复之后怎么防止下一次重演。文中会给 Kafka、RocketMQ、RabbitMQ 三套常见组合的监控命令和配置示例代码可以直接拿去改。本文适合正在维护线上消息中间件的后端开发和 SRE也适合正在做容量规划、想搞清楚到底该开几个消费者的同学。读完之后你至少能拿到三样东西一套可执行的堆积排查顺序、一份消费端提速的参数清单、一份常见故障对照表。1. 消息堆积问题速览先把关键结论摆出来。消息堆积的处理可以拆成三个层次止血、提速、治本。止血是让水位不再上涨甚至快速下降提速是让消费能力真正匹配生产速率治本是让系统在下次流量高峰时不再依赖人工介入。维度关键点说明核心指标Consumer Lag / 堆积条数 / 消费 TPS三个指标要一起看单看一个容易误判最快止血手段扩容消费者实例、提高并发度受分区数、队列数上限约束消费端提速批量拉取、批量处理、异步化、去掉同步 RPC收益通常比单纯加机器更大Broker 侧瓶颈分区数不足、磁盘 IO 打满、页缓存不足加消费者无效时优先怀疑这里顺序消息限制顺序消费无法靠加实例线性扩容只能扩分区或拆分业务 key典型误区只加机器不改消费逻辑消费逻辑里有同步阻塞时扩缩容收益极低恢复后动作补齐监控告警、压测消费上限、沉淀预案不做这一步大概率会二次堆积这里要强调一个判断原则当堆积量在涨、消费 TPS 却稳定不变时问题几乎一定在消费端或者下游依赖而不在消费者数量。反过来消费 TPS 随着实例增加而明显上升、但依然追不上生产速率才说明是消费者数量不够。2. 消息堆积的判定标准与常见根因2.1 什么程度才算堆积堆积本身不是故障短时间的 Lag 上涨是正常现象。真正需要介入的判断条件通常是下面几条里的任意一条Consumer Lag 持续上涨超过 5 分钟且没有回落趋势。堆积量对应的消费耗时已经超过业务可接受的延迟比如订单履约要求 3 分钟内处理完。消费 TPS 明显低于历史同期均值比如只有平时的三分之一。消息在 Broker 上的保存时间接近过期时间再不消费就会被清理。把这几条写进告警规则比单纯设置一个固定的堆积条数阈值更可靠因为不同 topic 的流量基线差别很大。2.2 生产端原因生产端突增是最常见也最容易识别的场景。典型表现是生产 TPS 曲线出现一个明显的台阶而消费 TPS 曲线几乎没动。常见诱因包括批量任务集中触发、上游系统重试风暴、某个业务线做数据回刷、定时任务时间点撞车。重试风暴值得单独说。当生产端对下游超时做了无退避的重试一次下游抖动会被放大成数倍的消息量而且这些消息内容往往高度相似消费端处理起来更容易触发同样的超时形成正反馈。处理这类问题的第一优先级不是加消费者而是给生产端加退避和重试上限。2.3 消费端原因消费端是堆积问题的主战场常见的几类消费逻辑里有同步的远程调用单条处理耗时被下游 RT 拖住。单条消息处理里有多次数据库单条写入没有做批量。消费线程池配置过小CPU 明明很闲但消费 TPS 上不去。消费者实例数已经等于分区数或队列数再加实例也分配不到分区。消费逻辑抛出异常导致无限重试某条消息反复占用消费线程。消费者频繁 Rebalance每次重平衡期间消费基本停摆。最后一条特别容易被忽略。日志里如果反复出现 rebalance 相关记录说明消费者在不停地进出消费组这时候无论怎么扩容都没用得先把会话超时、心跳间隔、单批处理耗时这几个参数调合理。2.4 Broker 与基础设施原因当消费端怎么看都没问题、加机器也没效果时要把视线转到 Broker分区数或队列数太少消费并行度被硬性限制。Broker 磁盘 IO 打满拉取请求响应变慢。页缓存不足导致大量读盘消费延迟抖动。网络带宽打满尤其是跨机房消费的场景。Broker 节点负载不均部分分区集中在热点节点上。这一类问题的特征是消费端的拉取耗时指标比如 poll 耗时、拉取 RT明显上升但业务处理耗时正常。3. 堆积排查的标准动作排查顺序很重要顺序错了会浪费大量时间。建议按下面这条链路走。3.1 第一步确认堆积范围先确认是单个 topic 堆积、单个消费组堆积还是全局堆积。这一步决定后面是查业务逻辑还是查基础设施。# Kafka查看指定消费组的堆积情况 kafka-consumer-groups.sh \ --bootstrap-server 127.0.0.1:9092 \ --describe \ --group order-consumer-group重点看输出里的LAG列和CONSUMER-ID列。如果某个分区的 LAG 特别大而其他分区很小说明是分区倾斜不是整体消费能力不足。# RocketMQ查看消费组进度 mqadmin consumerProgress -n 127.0.0.1:9876 -g order_consumer_groupRocketMQ 这里关注Diff Total和每个队列的Broker Offset与Consumer Offset差值。如果某个队列的 diff 远大于其他队列同样说明是队列级倾斜。# RabbitMQ查看队列堆积与消费者数量 rabbitmqctl list_queues name messages messages_ready messages_unacknowledged consumersRabbitMQ 要特别区分messages_ready和messages_unacknowledged。前者是待投递后者是已投递但没 ack。如果 unacked 很大而 ready 很小说明消费者拿到了消息但处理不完或者卡住了这时候加消费者往往有效如果 ready 很大而 consumers 数量正常说明投递或消费速度本身有问题。3.2 第二步对比生产与消费速率拿到堆积量之后立刻对比生产 TPS 和消费 TPS 的曲线。这一步能直接定性现象判断优先动作生产 TPS 突增消费 TPS 平稳流量突增扩容消费者 评估是否需要限流生产 TPS 平稳消费 TPS 下降消费端劣化查消费日志、下游依赖 RT生产 TPS 下降消费 TPS 也下降下游依赖故障查数据库、缓存、第三方接口两者都平稳但 Lag 在涨指标采集异常或分区分配异常查消费者实例与分区分配消费 TPS 为 0消费者掉线或阻塞查消费者进程、线程栈第三行是最容易被误判的。下游数据库慢查询会把消费线程全部占住表现就是消费 TPS 断崖式下跌同时因为业务失败生产端也在减少投递。3.3 第三步抓消费线程栈消费 TPS 异常时最快的定位手段是连续抓几次线程栈看消费线程卡在哪里。# 连续抓 3 次线程栈间隔 5 秒观察消费线程的停留位置 for i in 1 2 3; do jstack pid /tmp/consumer-stack-$i.txt sleep 5 done # 过滤出消费线程 grep -A 20 consume /tmp/consumer-stack-1.txt如果三次抓下来消费线程都停在同一个 JDBC 调用或者同一个 HTTP 客户端上问题就非常明确了下游依赖拖慢了消费。这时候加消费者的效果取决于下游能不能扛住更多并发盲目扩容有可能把数据库彻底压垮。4. 应急止血先把水位压下去堆积处理的第一原则是先让曲线掉头再谈优化。以下手段按见效速度排序。4.1 扩容消费者实例最直接的手段但有硬约束Kafka 的消费并行度上限是分区数RocketMQ 是队列数RabbitMQ 相对灵活但也有竞争消费者的开销。扩容前先确认当前实例数是否已经等于分区数。# 查看 topic 的分区数 kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --describe --topic order-topic如果实例数已经等于分区数扩容没有意义必须走后面的扩分区或者提升单实例消费能力路线。扩容时要注意滚动发布的节奏。一次性把所有消费者重启会触发多轮 Rebalance每轮重平衡期间消费基本停滞反而让堆积更严重。稳妥的做法是分批替换每批之间留出足够时间让消费组稳定。4.2 提高单实例并发度不改分区数的前提下提高单实例内部的消费并发是最快的提速手段。Kafka 通过concurrency控制RocketMQ 通过消费线程数控制RabbitMQ 通过并发消费者数量控制。Configuration public class KafkaConsumerConfig { Bean public ConcurrentKafkaListenerContainerFactoryString, String batchFactory( ConsumerFactoryString, String consumerFactory) { ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory(); factory.setConsumerFactory(consumerFactory); // 开启批量消费一次 poll 处理多条 factory.setBatchListener(true); // 单实例内的并发消费线程数实际生效值不超过分区数 factory.setConcurrency(8); // 手动提交避免批量处理中途失败导致位点错乱 factory.getContainerProperties() .setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); // 一次 poll 最多拉取的记录数 factory.getContainerProperties().setPollTimeout(3000); return factory; } }这里的concurrency不是越大越好。它本质上是单实例内启动的消费线程数线程太多会导致频繁的位点提交和上下文切换收益会递减。经验做法是先按分区数的一半设置观察消费 TPS 和 GC 情况再调整。4.3 开启批量消费批量消费是性价比最高的优化之一。单条消费的开销里网络往返、位点提交、事务开启占比很高把这些开销摊到一批消息上TPS 提升往往非常明显。Component public class OrderBatchConsumer { KafkaListener(topics order-topic, containerFactory batchFactory) public void onMessage(ListConsumerRecordString, String records, Acknowledgment ack) { ListOrderMessage orders new ArrayList(records.size()); for (ConsumerRecordString, String record : records) { try { orders.add(JSON.parseObject(record.value(), OrderMessage.class)); } catch (Exception e) { // 单条解析失败不要影响整批记录后跳过 log.warn(parse failed, offset{}, record.offset(), e); } } if (!orders.isEmpty()) { // 批量写库一条 SQL 搞定 orderMapper.batchInsert(orders); } // 整批处理完成后统一提交位点 ack.acknowledge(); } }批量消费有两个必须注意的点。一是批量大小要设上限比如max.poll.records控制在几百条量级太大容易触发 poll 超时被踢出消费组。二是失败处理要区分单条失败和整批失败单条解析失败不应该导致整批回滚重试。4.4 临时降级非核心逻辑堆积严重时临时把消费逻辑里的非核心步骤去掉是最快见效的手段之一。比如跳过实时推送、短信通知改为异步补偿。跳过明细日志落库只保留关键字段。关闭消费过程中的二次校验、风控查询。这类改动通常能在几分钟内把单条处理耗时降下来一半以上。前提是这些步骤有后续补偿机制不能直接丢。5. 消费端提速的深度优化止血之后进入提速阶段目标是让消费能力真正上一个台阶。5.1 把串行 RPC 改成异步或批量消费逻辑里最常见的性能杀手是一串串行远程调用。一条消息要查用户、查库存、查风控三次 RPC 加起来 60ms消费 TPS 就卡在几十。改成并行调用或者批量调用收益是直接的。// 串行写法三次 RPC 累加耗时 UserInfo user userClient.get(order.getUserId()); StockInfo stock stockClient.get(order.getSkuId()); RiskResult risk riskClient.check(order); // 并行写法耗时取决于最慢的那次调用 CompletableFutureUserInfo userFuture CompletableFuture.supplyAsync(() - userClient.get(order.getUserId()), executor); CompletableFutureStockInfo stockFuture CompletableFuture.supplyAsync(() - stockClient.get(order.getSkuId()), executor); CompletableFutureRiskResult riskFuture CompletableFuture.supplyAsync(() - riskClient.check(order), executor); CompletableFuture.allOf(userFuture, stockFuture, riskFuture).join();并行化要注意线程池隔离不能让消费线程和业务线程抢同一批资源。线程池大小按下游能承受的并发上限来定不要一味放大。5.2 消费幂等与去重提速过程中会引入重试和批量必然带来重复消费。幂等做不好提速反而会引入数据问题。public void handle(OrderMessage msg) { // 用消息唯一 ID 做去重设置合理的过期时间 String dedupKey mq:dedup: msg.getMsgId(); Boolean first redisTemplate.opsForValue() .setIfAbsent(dedupKey, 1, 24, TimeUnit.HOURS); if (Boolean.FALSE.equals(first)) { log.info(duplicate message ignored, msgId{}, msg.getMsgId()); return; } try { orderService.process(msg); } catch (Exception e) { // 处理失败要释放去重标记否则重试会被误判为重复 redisTemplate.delete(dedupKey); throw e; } }这里的细节是失败时必须删除去重标记否则下一次重试会被当成重复消息直接丢弃造成消息丢失。另一个可选方案是用数据库唯一索引兜底成本更低但依赖业务表设计。5.3 处理异常消息与无限重试一条反复失败的消息会长期占住一个消费线程在分区数少的时候影响非常明显。必须给重试设置上限。# Spring Kafka 重试与死信配置示例 spring: kafka: consumer: max-poll-records: 200 properties: max.poll.interval.ms: 300000 listener: ack-mode: manual_immediate配置层面之外业务代码里也要有兜底解析失败的消息直接进死信 topic业务处理失败超过 N 次的消息进死信并告警不要让它无限循环。死信消息要保留原始内容、失败原因和重试次数方便事后补偿。5.4 Rebalance 治理消费组频繁重平衡会让消费能力大幅波动。治理要点max.poll.interval.ms要大于单批消息的最长处理时间留足余量。session.timeout.ms和heartbeat.interval.ms保持合理比例通常心跳间隔是超时的三分之一。避免在消费线程里做长时间的阻塞操作。用静态成员static membership减少滚动发布时的重平衡次数。这一项的收益容易被低估。一个频繁重平衡的消费组实际有效消费时间可能只有一半。6. Broker 侧与架构层面的治理消费端优化到头之后剩下的瓶颈通常在 Broker 侧或者分区设计上。6.1 扩分区与扩队列分区数决定了 Kafka 消费并行度的天花板队列数决定了 RocketMQ 的上限。扩分区能直接提高并行度但有几个副作用需要提前评估扩分区会打乱原有消息的 key 到分区映射依赖同一 key 有序的场景会失效。扩分区后历史消息的分区分布不会改变新分区只承接新消息短期内可能出现负载不均。RocketMQ 扩队列后需要重新分配消费队列过程中可能有短暂延迟。所以扩分区属于计划内操作不适合作为应急手段。应急阶段更现实的是把单实例消费能力提上去。6.2 顺序消息的特殊处理顺序消息是堆积治理里的难点。顺序消费要求同一个业务 key 的消息落在同一个分区或队列并行度天然受限。这类场景的处理思路不是加并发而是缩小顺序粒度把全局顺序改成按用户 ID 或订单 ID 顺序让不同 key 之间可以并行。把大 key 拆开避免单个热点用户把整个队列堵住。对热点 key 单独建 topic隔离影响。如果业务确实需要全局顺序那只能在单条处理耗时上做优化并行度这条路基本走不通。6.3 限流与降级有时候正确的做法不是追着消费能力跑而是在生产端限流。当消费端已经满负荷、下游数据库也接近极限时继续把消息灌进来只会让整个链路雪崩。限流的落点可以在生产端、Broker 端或者消费端入口。生产端限流对业务侵入最大但最有效Broker 端可以做配额限制消费端入口限流能保护下游但会让堆积继续增长。具体选哪一层取决于堆积的容忍度和下游的保护需求。7. 接口与监控集成示例堆积治理离不开监控。下面给一套通用的监控采集与告警思路具体接口路径需要按实际使用的中间件和管理平台调整。import time import requests # 通用模板从监控接口拉取消费组堆积数据 # 实际 URL、鉴权方式和返回结构需要按所用中间件的管理接口调整 def fetch_lag(group_id: str) - dict: url http://127.0.0.1:8080/api/consumer/lag params {group: group_id} resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json() def check_and_alert(group_id: str, threshold: int) - None: data fetch_lag(group_id) lag data.get(totalLag, 0) consumer_count data.get(consumerCount, 0) if lag threshold: print(f[ALERT] group{group_id} lag{lag} consumers{consumer_count}) if __name__ __main__: while True: check_and_alert(order-consumer-group, threshold100000) time.sleep(60)告警规则建议分层设计告警级别触发条件处理方式提醒Lag 超过基线 2 倍持续 5 分钟值班群通知观察趋势警告Lag 持续上涨 10 分钟无回落排查消费端准备扩容严重消费 TPS 归零或 Lag 超过 1 小时处理量立即介入启动预案致命消息临近过期时间仍未消费评估数据丢失风险升级处理分层的意义在于避免告警疲劳。所有堆积都报严重值班同学很快就会对告警脱敏。8. 常见问题与排查方法问题现象可能原因排查方式解决方案扩容消费者后 Lag 不降实例数已达分区/队列数上限查看 topic 分区数与消费者实例数扩分区或提升单实例消费能力消费 TPS 周期性掉零频繁 Rebalance查消费组日志中的 rebalance 记录调整会话超时、分批发布、启用静态成员消费线程全部阻塞下游 RPC 或数据库慢连续抓线程栈看停留位置加超时、并行化、降级非核心逻辑单条消息反复重试消息体异常或业务逻辑缺陷查死信队列和消费异常日志设置重试上限异常消息进死信并告警unacked 消息持续增长消费者处理不完或未正确 ack对比 ready 与 unacked 数量检查 ack 逻辑提高并发消费端 CPU 不高但 TPS 上不去消费逻辑 IO 密集或线程池过小看线程池活跃数与队列积压调大消费线程数改批量处理Lag 指标与实际不符监控采集延迟或位点提交异常用命令行工具交叉验证修正采集逻辑统一统计口径消息临近过期仍未消费堆积时间超过保留期对比最早未消费消息的时间戳临时延长保留期优先消费旧消息部分分区 Lag 特别大分区倾斜或热点 key按分区看 Lag 分布检查 key 设计必要时拆分热点消费速度忽快忽慢Broker 磁盘 IO 抖动或页缓存不足看 Broker 磁盘指标与拉取 RT扩容 Broker、优化磁盘、增加分区这张表里的每一条都建议对应到具体的监控项和操作手册。堆积处理最怕的是现场临时想方案有预案和没预案恢复时间能差好几倍。9. 最佳实践与长期治理堆积治理做完一次不算完要把经验沉淀下来。容量规划要留余量。消费能力的规划设计值不要贴着日常峰值走至少要能扛住日常峰值的两到三倍。分区数同样要留出扩容空间一次规划到位比事后扩分区成本低得多。消费逻辑要有明确的耗时预算。给每类消息定一个处理耗时上限超过就告警。这样消费端劣化能在演变成堆积之前被发现而不是等到 Lag 涨到十万条才有人注意到。批量、异步、幂等三件事要成套做。只做批量不做幂等重试会带来脏数据只做异步不做超时控制线程池会被慢慢耗尽。三件事配套落地提速才是安全的。死信队列必须有人看。死信队列不是垃圾桶进了死信的消息等于业务丢了。要有定期的死信巡检和补偿流程否则堆积问题只是被转移了没有真正解决。定期做消费能力压测。每隔一段时间人为把消费端限流到某个值观察堆积上涨的速度和恢复的速度验证告警规则和扩容预案是否还有效。这套演练做一次比写十页文档管用。保留最小可运行配置。消费端的并发数、批量大小、超时时间这些参数改来改去很容易失控。维护一份基准配置并记录每次调整的原因和效果下次遇到同类问题可以直接参考。10. 收尾从最小验证开始如果你现在手里正好有一个堆积的消费组建议按这个顺序动手先用命令行工具确认堆积分布是整体还是分区级再抓两次线程栈看消费线程卡在哪确认是消费端问题之后优先改批量消费和并发数这两项改动风险最低、见效最快最后再评估是否需要扩分区或者调整架构。最容易踩的坑有三个。一是扩容前没看分区数上限加了一堆实例结果分配不到分区白折腾。二是只改消费者不动消费逻辑线程加了一倍 TPS 只涨了百分之十。三是堆积恢复之后不补监控不写预案下次同样的流量高峰再来一遍。真正值得先验证的功能其实就三个能不能准确看到每个分区的 Lag、能不能在一分钟内把消费并发调上去、异常消息能不能自动进死信而不是无限重试。这三件事跑通了大部分堆积场景都能在可控时间内恢复。
阅读完成 · 觉得有帮助？
