凌晨两点被电话叫醒说线上订单消息积压了上百万条下游清算跑不动用户退款一直卡在“处理中”。我爬起来打开监控面板消费者组 Lag 曲线像垂直起飞一样往上拉消费速率几乎归零而生产端还在源源不断地写入。这不是我第一次处理 MQ 消息积压但每次的根因都不太一样有的是消费线程被远程调用卡死有的是重平衡频繁触发有的竟然是消费逻辑里不小心写了个慢 SQL。这篇文章就是想把这类问题的排查思路和优化手段系统梳理一遍从现象定位、根因分析到消费速度优化都是我在实际环境里验证过的做法适合正在跟消息堆积死磕的后端开发、运维和架构师参考。消息积压这件事看起来只是“消费速度跟不上生产速度”但背后的原因可能藏在生产者、Broker、消费者任何一个环节里。如果你不按套路出牌上来就盲目加大消费者线程数大概率治标不治本甚至可能把下游数据库打挂。下面我按自己的排查习惯从定位开始一步步讲。1. 消息积压的定位从现象到根因1.1 积压是怎么产生的先分清是“生产太快”还是“消费太慢”积压的时间点很重要。如果积压是从某个时间点突然开始的往前推那个时间点通常就是事故现场。我见过最典型的情况是上线了新功能把一条原本轻量的消息体里塞了一个巨大的 JSON 字段消费端每次反序列化要多花几十毫秒量一大直接原地崩溃。所以排查第一步不是看消费者而是先看积压曲线的斜率。如果积压是“匀速增长”的说明消费速率稳定地低于生产速率这时候重点找消费端吞吐的瓶颈。如果积压是“垂直起飞”的说明消费端在某一个瞬间几乎完全停摆优先怀疑消费者线程卡死、频繁重平衡、下游依赖超时或者消费进程本身在疯狂 GC。区分这两个场景有个很实用的办法看消费者的“已消费消息数”指标。如果消费数量还在涨但涨得很慢是消费能力不足如果消费数量完全不动了那就是消费卡死。这一步判断对了后面排查路径基本就对了。另外还要看一眼生产端有没有异常突刺。比如某个定时任务集中触达或者某个大促活动整点放量生产速率瞬间翻了十倍。这种情况消费者其实没毛病只是瞬时流量超过了消费能力处理思路是削峰填谷和限流而不是玩命调消费者参数。1.2 快速定位积压的命令与看板指标不同的 MQ 产品看积压的姿势不一样但思路是通用的看消费进度Offset / Delivery Tag和生产/消费速率差。Kafka 环境下最直接的方式是用命令行工具查消费者组的 Lagkafka-consumer-groups.sh --bootstrap-server your-broker:9092 \ --describe --group your-consumer-group输出里会列出每个 Topic 分区对应的 Current-Offset 和 Log-End-Offset两者的差值就是 Lag。如果某个分区的 Lag 特别大而其他分区都很正常那么大概率是分区分配不均匀或者某个分区对应的消费者实例已经挂了。RocketMQ 这边可以用mqadmin工具查看消费进度和积压量mqadmin consumerProgress -n your-name-server:9876 -g your-consumer-group这个命令会直接显示每个 Topic 的消费堆积总量和消费速率信息非常直觉。RabbitMQ 的管理界面本身就带 Queue 的 Ready / Unacked 数量就是积压量。注意要区分这两个数Ready 是等着被消费的消息Unacked 是已经发给消费者但还没确认的消息。如果 Unacked 数量持续很高说明消费者明明拿了消息却一直不 ack大概率是消费逻辑卡住了。看板层面我建议至少盯四个指标积压数量Lag / Ready / Unacked消费速率每秒成功消费并 ack 的消息数消费失败率消费者实例存活数这四个指标任何一个异常都能快速缩小排查范围。没有监控系统的团队至少要写个脚本定时把这些指标丢到日志或者告警平台里别等线上报警了才去翻命令。2. 消费卡顿的深挖不要让表象骗了你2.1 会阻塞消费线程的几个隐形陷阱消费者卡顿最常见的原因是线程阻塞在外部依赖上而且很多时候不容易一眼看出来。我常用的排查手段是在消费消息的入口和出口分别打耗时日志统计单条消息处理的 P99 和 P999。如果 P999 明显偏高下一步就用 Arthas 的thread命令看消费线程到底卡在哪个调用栈上。先说数据库连接池耗尽。很多消费逻辑里会写数据库从连接池拿不到连接时线程会阻塞等待而不是立刻失败。如果连接池最大连接数配的是 20而下游数据库的活跃连接已经满了消费线程就会跟着一起排队。配合数据库侧的show processlist能看见大量 Sleep 或 Waiting for table lock 的会话。这种情况光调 MQ 没用得把连接池参数和慢 SQL 一起治理。再说说分布式锁。我之前踩过一个坑消费者处理消息时要执行一段业务逻辑为了防并发先抢一把基于数据库的分布式锁。结果锁的持有时间被一个外部接口拖到了几十秒导致后续消息全部阻塞等待积压一路飙升。排查时系统资源看起来都很正常就是消费一直不前进。后来优化思路是尽量缩短持锁时间锁内只做核心判断重操作放到锁外执行。第三个容易忽略的是序列化性能。比如用 JSON 解析一个嵌套很深的大对象或者用默认 JDK 序列化处理二进制数据这些操作在低并发下没啥感觉但消息量一旦上来就是灾难。遇到这种问题优先考虑换成 Protobuf、MessagePack 这类高效序列化方案或者干脆改造消息体结构把大字段拆出来存到独立的存储里消息里只放引用 ID。2.2 依赖超时与线程池耗尽最常见的卡顿元凶远程调用超时导致的消费阻塞在所有积压事故里能占三成以上。最典型的场景消费消息时要调一个下游 HTTP 接口代码里没设置connectTimeout和readTimeout使用的是 HTTP 客户端默认值有的默认是 0即无限等待。下游一抖动消费线程全部挂住积压量直线上升然后下游一恢复这些线程又全部涌入直接把下游再次打挂。这种雪崩式排队在线上我见过不止一次。正确的处理姿势是所有消费链路里的远程调用都必须设置超时而且超时时间要小于 MQ 的消费超时时间。Kafka 里有max.poll.interval.ms如果消费者两次 poll 的间隔超过这个时间就会被认为是死亡触发重平衡。RocketMQ 里有消息消费超时时间超过后消息会被重新投递。也就是说消费线程阻塞的时间如果超过了这些阈值还会引发消息重复投递和重平衡把问题进一步放大。线程池耗尽也是一个常见元凶。有些团队喜欢在消费逻辑里再丢到一个自定义线程池里做异步处理如果线程池用了无界队列积压任务会全都堆在内存里如果线程池不够大消费线程拿到消息后都去提交任务然后再去拿新消息看起来消费速率正常实际业务处理根本没跟上积压数据照样在涨。这种设计不是不行但一定要看清线程池的拒绝策略和队列上限绝不能无脑提交。还有一个隐蔽的卡顿源是 GC。如果消费进程的堆内存配置过小或者消息体过大导致频繁创建大对象可能触发频繁的 Full GC整个 JVM 停顿几秒甚至十几秒。期间消费者自然无法处理消息积压就上来了。排查时注意看 GC 日志和 CPU 使用率如果 CPU 不高但应用无响应大概率是 GC 停顿。3. 消费速度优化的实战手段3.1 调整消费并发与分区负载消费并发和 MQ 的分区模型强相关不是说你起十个消费者线程速度就能提高十倍。Kafka 的并发上限由分区数决定。一个分区只能被同一个消费者组里的一个消费者实例消费所以消费者线程数并行上限就是分区总数。如果你的 Topic 只有三个分区起了十个消费者实例其中七个会闲着。这种场景的优化路径是评估每个分区的单线程消费能力结合目标吞吐先把分区数扩容到合理值比如目标每秒 5000 条单线程每秒处理 200 条至少 25 个分区再把消费者实例数调整到接近分区数。RocketMQ 这边略有不同它支持一个消费者实例内启动多个消费线程consumeThreadMin和consumeThreadMax参数可以控制线程数。RocketMQ 的默认值是 20如果你的消费逻辑偏 IO 密集这个值可以适当调大比如 40 到 60如果消费逻辑是纯 CPU 计算线程数超过 CPU 核心数反而会增加上下文切换开销。RabbitMQ 的并发优化方式和前面两个不太一样。RabbitMQ 基础模型里没有分区的概念但是消费者可以设置prefetchQoS和并发消费者数量。prefetch表示每个消费者同时持有的未确认消息数默认值是 250。这个值如果太大消费者会一次性拉取大量消息到本地内存处理不过来就全部变 Unacked其他消费者也分不到消息。一般建议根据单条处理时间和吞吐目标来调整处理快的可以开 100 到 200处理慢的降到 10 到 50。3.2 批量消费与会话参数调优批量消费是提升吞吐最直接的手段之一但不同消息队列的批量方式不同。Kafka 本身是批量拉取的关键参数是max.poll.records也就是单次 poll 最多返回多少条。默认值是 500如果每条消息处理很快可以调大到 1000 甚至 5000减少 poll 的开销和上下文切换。但这个值不是越大越好max.poll.interval.ms限制了从 poll 到下一次 poll 的最大间隔如果一次 poll 回来的 5000 条消息处理时间超过了这个限制消费者会被判定死亡触发重平衡。所以调大max.poll.records的同时要同步调大max.poll.interval.ms或者把批量处理逻辑改成分批。RocketMQ 支持消费端一次拉取多条消息consumeMessageBatchMaxSize默认是 1。把它调成 10 或 20配合循环调用MessageListener中的批量处理逻辑能够明显提升消费吞吐。这里有一个经验不要死板地把批量消息一条条处理尽量利用资源的局部性比如一次批量处理里批量更新数据库而不是逐条 update。RabbitMQ 的批量消费需要自己在消费者里做缓冲。比如利用basicConsume不断接收消息攒够 100 条或者每隔 100 毫秒再批量处理一次使用manual ack处理成功后再统一 ack。这样做法的好处是减少单条 ack 的网络开销坏处是如果消费者进程崩溃缓冲区内未 ack 的消息会全部重新投递对下游的幂等性要求更高。另外一个值得关注的参数是消费端的网络和会话超时。Kafka 的session.timeout.ms默认在 45 秒左右如果消费线程频繁 GC 导致心跳发送延迟就会触发重平衡。可以适当调大这个值来减少误判但别调得太大否则消费者真的挂了也没人知道。3.3 消费逻辑精简与异步化消费速度上不去很多时候不是 MQ 配置的问题而是业务逻辑太重。你需要像一个过滤器一样审视消费链路把非核心的步骤从消费主链路里剥离。我见过一个订单同步消息的消费逻辑里面依次做了查订单、查用户、查商品、调库存接口、发短信通知、推送实时数据、更新订单状态、记录操作日志。一条消息从头到尾要两秒钟积压不涨才怪。后来拆分成了两个消费者主链路只做订单状态同步然后发一条“派生消息”给另一个 Topic由一个独立的消费者去处理短信、推送、日志这类非核心动作。主链路耗时降到了几十毫秒积压问题迎刃而解。异步化也是同样的思路。比如消息里需要写入数据库不要逐条写而是先缓存到内存队列批量 flush 到数据库。比如消费时要把一条消息发给多个下游系统可以并行调用而不是串行调用使用CompletableFuture或者自行管理线程池把总耗时从“所有下游耗时之和”变成“所有下游耗时最大值”。还要注意不要在消费线程里做重计算。比如需要对消息做数据校验、字段映射、格式转换这些如果耗时明显就考虑预计算或者旁路缓存。我当时做过一个优化消费时要把用户 ID 反查成用户等级每次查一次数据库后来改成每天把用户等级加载到本地缓存里消费时直接内存查询吞吐直接翻了三四倍。4. 从消费端根治积压全链路优化4.1 消费幂等与重复消息处理只要聊到消费速度优化就绕不开幂等这个话题。原因很简单当你为了提高消费吞吐而开启批量消费、异步确认、失败重试的时候消息重复投递的概率会同步上升。如果你没有做好幂等重复消息会对下游产生脏数据甚至反向拖垮数据库最终还是会积压。先明确一个事实几乎所有主流 MQ 都不能保证绝对不重复只提供“至少一次”或“至多一次”的语义。Kafka 开启幂等生产者只能保证生产者端不重写重复消息但消费者端的网络抖动、重平衡、事务回滚照样会导致重复投递。RocketMQ 的默认投递语义也是 at-least-once。所以消费端的幂等是刚需不是附加项。网上讨论很多的一个问题是前端点两次算是发两条消息吗从消息队列的角度看如果前端把两次点击各自当成一次请求发给后端后端又各自发了一条 MQ 消息那 MQ 不背这个锅它就是收到了两条不同的消息。这种情况的治理应该从源头入手前端做防抖后端做去重比如基于用户操作 ID 做幂等。如果你指的是同一笔业务操作被前端重复提交后端在入口用分布式锁或数据库唯一约束挡住只允许第一次请求产生 MQ 消息那 MQ 就不会有重复消息。消费端幂等有几种常见方案我按推荐程度排序说说第一种是业务唯一键去重。消息里带一个业务唯一 ID比如订单号、支付流水号。消费的时候先去 Redis 查一下这个 ID 是否处理过没有则处理并写入 Redis有则直接丢弃。这种方案的关键是 Redis 的写入和处理逻辑要保证原子性建议使用SETNX或者 Lua 脚本避免并发下两个消费线程同时查到“未处理”然后各执行一遍。第二种是数据库唯一约束。如果消费的落库表里有自然唯一键比如订单号 消息类型直接建唯一索引重复插入会报错捕获异常后忽略即可。这个方案的优点是天然可靠不用关心 Redis 里 key 过期的问题缺点是需要改表结构和业务逻辑。第三种是状态机控制。比如订单状态从“待支付”到“已支付”只能流转一次如果收到重复的“已支付”消息发现当前状态已经是“已支付”直接跳过。这个方案特别适合业务本身有状态流转的场景实现简单且符合业务语义。幂等的核心原则是不要在消费逻辑里假设每条消息只来一次而是假设每条消息可能来无数次然后从业务上保证后一次处理不会对前一次的结果产生破坏。4.2 监控、告警与积压预防机制很多团队处理消息积压都是等报警响了才四处排查。但真正成熟的方案是提前把“积压风险”扼杀在摇篮里。这里分享几个我自己在用的监控和预防经验。首先积压告警不能只设置一个固定阈值。比如你 Topic 正常 Lag 在几百设置一万才告警可能等你看到告警的时候已经积压了几分钟补救成本很大。我的经验是分三级轻度积压比如持续一分钟内 Lag 超过正常值的三倍通知开发确认中度积压持续五分钟以上触发告警和自动扩容流程重度积压影响接口响应或下游链路直接电话唤醒。其次关注消费速率的“斜率”比关注绝对值更有提前量。绝对值只能告诉你现状斜率能告诉你趋势。比如这一分钟内 Lag 涨了 5000 条但上一分钟只涨了 1000 条说明消费速度正在恶化这时候即使绝对值还不高也要提前介入。然后自动化扩容机制值得认真做。Kafka 场景下如果检测到消费者组整体 Lag 持续升高可以自动把消费者实例扩容。前提是你的消费者是以集群方式部署的且分区数足够。RocketMQ 场景下可以自动调整消费线程池大小但要注意线程数上限和下游承受能力。最后一定要对生产端的流量做管控。很多积压其实是上游一次性灌入过多数据导致的。比如定时任务扫表补偿、数据迁移脚本、大促活动瞬间放量。这类上游流量最好做限流或者分桶削峰别让 MQ 变成流量冲击的缓冲池。MQ 本身不是数据库它不适合长期保存大量积压消息。另外提一句不要迷信“MQ 官网”或者任何单一信息来源。每个团队的 MQ 版本和部署方式都有差异官方文档只能指导通用用法线上环境的真实瓶颈永远要靠自己的监控数据和日志来验证。多收集自己环境里的基线数据比反复查文档有用得多。5. 实测案例与问题排查实录5.1 一次订单积压事故的完整复盘去年我处理过一起比较典型的积压事故花了一个多小时才定位到根因。复盘出来其实是一个很常见的组合问题分享给大家当案例。现象订单系统的一个消费者组 Lag 快速上升从几千涨到几十万消费速率趋近于零。第一反应是扩展消费者实例加了三台机器之后Lag 没有任何下降趋势。排查过程第一步看消费者日志发现大量消息处理抛异常异常类型是数据库锁等待超时。这说明卡点在下游数据库而不是 MQ 本身。但奇怪的是数据库的 CPU 和连接数都不高。第二步继续看异常堆栈发现等待的是某张表上的行锁。顺藤摸瓜查这张表发现有一个批量补偿任务在全表扫描更新数据持锁时间很长把正常消息处理要更新的行全部堵住了。第三步看消息内容发现这批消息改了一个新字段而对应的数据表加了新索引导致消费逻辑里的 update 语句走错索引行锁范围扩大锁冲突概率飙升。最终处理分三步走暂停批量补偿任务让正常消费先跑完把新索引调整成联合索引减少行锁范围对消费逻辑里的 update 增加重试机制碰到死锁或锁等待时自动重试三次。这个案例的典型性在于问题根子根本不在 MQ而在下游数据库。如果你只看 MQ 的积压曲线会觉得消费者拉胯实际是下游不给力。排查消息积压一定要把消费链路当成一个整体来看从消息拉取到业务处理再到最终落库任何一个环节都能成为积压的源头。5.2 常见问题速查表从症状到对策排查多了之后我习惯把常见问题收敛成一张速查表每次遇到积压告警就先对照一遍能省下不少时间。这里公开出来供大家参考。症状可能原因排查手段对策Lag 垂直起飞消费速率归零消费者线程卡死 / 重平衡 / JVM 停顿看消费线程堆栈、GC日志、Consumer 心跳状态定位阻塞点设置超时调 GC 参数Lag 匀速增长消费速率稳定偏低消费逻辑太重 / 并发不足统计单条消息耗时分布拆分耗时步骤异步化、批量消费、增加分区和实例单分区 Lag 异常高分区分配不均 / 单实例消费者卡顿查看分区分配情况和各实例消费进度均衡消费者实例处理单实例阻塞Unacked 消息堆积RabbitMQ消费者拿了消息不 ack看消费者线程状态和业务处理耗时调整 prefetch修复消费逻辑消费报错但 MQ 无异常下游数据库/接口异常看消费日志和下游系统监控增加重试、熔断治理下游性能消息重复消费引发脏数据消费端缺少幂等核对消息唯一ID和业务落库情况加唯一约束或 Redis 去重这张表适合贴在团队文档里作为积压排障的第一份参考。实际排查时建议严格按照“先看监控曲线 - 再看消费日志 - 再看下游系统 - 最后调参数”的顺序不要跳步。5.3 压测验证调优结果怎么才算有效优化消费速度不能凭感觉。参数调完必须用压测或者回放验证效果否则你根本不知道改动是正向还是负向。我的做法是准备一套压测环境用真实消息流量脱敏后灌入 MQ分别记录调优前后的消费速率和 Lag 曲线。重点关注三个指标最大消费速率每秒处理并 ack 的消息数单条消息平均耗时和 P99持续高吞吐下是否出现消费失败或重复投递比如把max.poll.records从 500 调到 5000压测发现单次 poll 处理时间变长触发了max.poll.interval.ms超时说明这个值对当前消费逻辑来说太大了需要回调到 2000 或者调整max.poll.interval.ms。批量消费从 1 调到 20 时观察下游数据库的写入延迟和连接池占用。如果数据库连接池出现排队等待说明批量带来的收益被数据库瓶颈抵消了需要同步扩容数据库连接池或优化 SQL。压测还有一个容易踩的坑压测用的消息内容如果太短序列化和反序列化的耗时会被低估如果消息字段太复杂又可能高估。最好直接拿一段线上真实消息体脱敏后作为压测样本这样结果才有参考意义。写在最后排查积压的几个心得消息积压排查做了这么多年最大的体会是越急越容易走弯路。告警一响人很容易慌上来就加消费者实例、调并发参数结果往往只是把矛盾转移到了下游甚至引发更严重的连锁故障。正确的姿势是先花几分钟把积压曲线、消费速率、异常日志看明白确定瓶颈究竟在研究哪个环节再动手优化。另外一个很实用的习惯是把每次积压事故的根因和解决过程记录下来沉淀成排查手册。同一套系统里积压的根因往往不会只有一种但排障路径是相似的。有了手册团队里的新同学也能在关键时刻顶上。如果你现在正好在加班排查积压建议先做一件事打开监控看看 Lag 是匀速增长还是垂直起飞。这个答案能帮你至少省掉一半的排查时间。
)
