前几天凌晨一点我被一条告警电话叫醒订单消息积压了十几万条消费者的日志还在刷但消费速度就是提不上去。类似这种MQ消息积压的问题相信很多做后端的同学都遇到过。消息积压、消费卡顿、消费速度上不去这三个词凑在一起往往意味着线上交易快断了。今天就把我排查这类问题的完整思路和方法整理出来希望能帮大家少走一些弯路。这篇文章适合谁看主要是有一定MQ使用经验RabbitMQ、Kafka、RocketMQ都适用但对性能调优和故障排查还不太系统的后端开发者。我会尽量从底层原理讲到操作命令最后再分享一些实战中踩过的坑。无论你是正在处理线上告警还是想提前预防消息堆积都可以参考这套方法论。注意我下面说的都是通用思路具体参数和命令以你使用的MQ版本、云厂商控制台为准。如果只是“刚学MQ、想知道怎么在页面查看消息”的读者建议先看第二部分那里有比较直观的页面操作说明。1. 先搞清楚消息积压是怎么发生的1.1 一条消息从生产到消费到底经历了什么要排查积压首先得清楚一条消息的完整生命周期。以最常见的生产-消费模型来说消息大致要经过生产者发送到BrokerBroker持久化存储消费者从Broker拉取RabbitMQ中也有Push模式本质还是Broker推给消费者业务逻辑处理最后向Broker确认消费完成。这里最关键的是“确认”这一步。RabbitMQ只要消费者返回BasicAck消息才会被真正标记为已消费Kafka是消费者提交offsetoffset之后的消息才认为处理完成RocketMQ则是消费成功后返回CONSUME_SUCCESS。如果哪一步卡住了消息就会停留在“已投递未确认”或“未提交offset”的状态而新消息还在不断进来积压自然越来越严重。我看到不少同学一看到消息积压第一反应就是“生产者发太快了”整天盯着生产端调限流。但实际排查时大部分积压问题都出在消费端要么消费线程卡死要么处理逻辑太慢要么参数配置不合理。生产者速率只是导火索不是根因。1.2 积压的本质生产速度长期大于消费速度我用一个打饭的例子解释积压食堂大师傅每秒能炒两个菜生产速率打饭阿姨每秒只能服务一个同学消费速率那么每秒都会有一个同学排队等着队伍越排越长。MQ堆积就是这个队伍的长度。如果只是偶尔一瞬间生产方突然涌进来一批消息消费方可能很快消化掉这叫瞬时积压。但如果你看到积压量持续增长比如每过一分钟多一千条那就要警惕了要么生产速率确实长期高于消费速率要么消费速率正在下降更隐蔽。我在定位问题时习惯先看两个指标积压数量曲线和消费速率曲线。积压上涨的同时消费速率如果是平的那就是生产能力问题如果消费速率明显下降那就是消费侧出故障了。可以简单套用公式估算积压增长趋势积压量 (生产速率 - 消费速率) × 持续时长。这个公式虽然不精确但能帮你判断问题严重程度。比如积压10万条消费速率500条/秒理论上需要200秒才能清完如果业务可以接受就不必过度惊慌如果不能接受就得马上扩容或者优化消费逻辑。2. 排查消息积压的第一步确认积压数据与位置2.1 从控制台/命令行确认积压数量不同MQ查看积压的方式不太一样先说最快的方式看控制台或执行命令行。如果你的环境是RabbitMQ可以使用rabbitmqctl命令rabbitmqctl list_queues name messages messages_ready messages_unacked这条命令会输出每个队列的当前总消息数、等待投递Ready的消息数、以及已经投递给消费者但还没确认Unacked的消息数。注意如果messages_ready很高说明消费者根本没来得及拉取如果messages_unacked也非常高说明消息已经拉走了但消费者处理不动彻底卡住了。这两种情况原因不同排查方向也不同。如果是Kafka最常用的是查看消费组LAGbin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group your_group输出结果里每一行代表一个分区LAG列显示的是该分区还积压多少条消息。Kafka的LAG是按分区统计的如果某个分区的LAG特别高而其他分区正常说明这个分区的消费者可能出问题或者分区数据倾斜。RocketMQ则可以用mqadmin命令查看消费进度mqadmin consumerProgress -g your_group它会列出该消费组下所有Topic的消费进度包括Broker中的消息总量和已经消费位点两者相减就能得到积压数量。也可以直接打开RocketMQ Dashboard在“消费者”页面看Diff总量。拿到积压数据后建议先记录一下时间点和数量过几分钟再看一次确认积压是在增长、持平还是回落。增长说明消费跟不上持平说明双方暂时均衡但总量已经很高了回落则意味着系统正在恢复。2.2 在页面上查看消息的几个实用入口很多刚接触MQ的同学经常会问mq怎么在页面查看消息其实不同MQ差别挺大。RabbitMQ只要开启了Management插件浏览器访问15672端口进入Queues页面点击队列名就能看到Messages、Message rates等指标。如果要看消息体内容可以在队列页面底部的“Get messages”区域填入数量点击Get Message按钮获取。这里要特别提醒获取出来的消息默认会被消费移除如果你只是想看一眼不想破坏消息必须把Ack Mode改成“Nack message requeue true”这样消息取出来后还会塞回队列。Kafka本身没有官方网页控制台生产环境一般通过第三方界面工具比如Kafka UI、Kafdrop来查看。这类工具通常也支持查看消费组LAG、浏览消息内容。如果你用的是云厂商的托管Kafka一般云控制台里也会提供“消息查询”功能支持按照Topic、分区、时间范围或者消息Key检索消息。RocketMQ Dashboard算是最直观的在“消息”页签中你可以按照Topic、Message ID、Key三种方式查询消息详情。我在排查消费卡顿时经常先用Message ID反查消息的消费轨迹看它是否被消费者拉走以及消费者返回的结果是什么。如果消息一直显示“消费中”多半是消费者线程处理到时卡住了。2.3 顺着消费位点定位卡点只看积压总数不够还要定位消息到底卡在哪个环节。以Kafka为例查看消费组详情时除了LAG还会显示CURRENT-OFFSET和LOG-END-OFFSET。如果CURRENT-OFFSET长时间不动说明消费者可能已经停止消费了不一定是宕机也可能是线程阻塞了。这时候要用jstack看一下消费者进程的线程栈找到那些处于WAITING、BLOCKED状态的消费线程。RabbitMQ则要看Unacked数量。正常情况下Unacked会稳定在一个小范围内如果它始终居高不下且Ready接近0说明消息全在消费者手里但迟迟不Ack。这种一般不是网络问题而是业务处理超时或者回调逻辑有bug。3. 消费卡顿的常见原因与定位方法3.1 消费者线程数配置不当这是最容易被忽视的原因。很多同学使用Spring Boot集成RabbitMQ时直接在方法上加了RabbitListener但没配置concurrency默认的并发消费者线程数只有1。一条消息处理2秒那么吞吐就是0.5条/秒一旦消息量稍微上来瞬间积压几十万。类似的Kafka的普通消费者如果不手动设置线程池往往是在一个循环里poll然后处理如果你在poll循环里写的业务逻辑很重整个消费就变成串行处理了。Kafka本身不支持为一个分区启动多个线程并发消费因为会破坏分区内消息的顺序但你可以通过增加分区数再增加消费者实例来提升并行度。我见很多团队上线后从不关注消费者并发配置等到积压告警才慌。这里给一个初始参考值如果单条消息处理耗时为T秒希望达到Q条/秒那么至少需要Q×T个并发消费者/线程。比如单条消息耗时0.2秒目标每秒处理500条至少需要100个并发线程。当然这只是理论下限实际还要考虑数据库连接、CPU、GC等资源开销。3.2 业务处理逻辑耗时过长或依赖外部服务超时消费卡顿最常见的原因不是M Q本身慢而是消费者里的业务逻辑变慢了。我排查过好几个项目发现消费性能瓶颈几乎全在“消费消息时调用其他系统”这一步比如消费订单消息时去调库存服务库存服务如果响应慢或者超时时间设置过长消费线程就会一直阻塞等待。曾经遇到一个事故消费端调用下游物流接口HTTP客户端默认超时时间是30秒结果下游服务被大流量打挂响应变得极慢半分钟内不返回消费线程全部卡在IO等待上一条消息处理超过30秒消息越堆越多。后来在日志里加入耗时统计才发现很多消费耗时都集中在外部调用上。所以排查时一定要看消费日志里“单条消息处理时长”的分位数如果P99明显高于平时优先检查慢SQL、外部接口、Redis操作。外部调用一定要设置超时和熔断。我常用的原则是MQ消息消费里的外部RPC超时时间不要超过3秒而且必须配合快速失败。与其让线程卡在外部服务上不如把消息重新入队或者投递到延迟队列下次再试。3.3 连接/通道/拉取模型配置不合理这个属于MQ消费参数调优的范畴。RabbitMQ有个关键参数prefetch表示消费者在确认完消息之前能预取多少条消息到本地。Spring Boot默认的prefetch是250看起来大其实隐患很大如果消费者处理速度慢250条消息全部拉给这个消费者了其他消费者就只能空等而且这250条消息都处于Unacked状态一旦消费者崩溃还要重新入队。对于慢消费者prefetch设成1到10反而更合理可以保证多个消费者平均分配消息。Kafka那边对应的是max.poll.records和max.poll.interval.ms。max.poll.records设置太大一次poll拉几百条处理时间太长超过max.poll.interval.ms默认5分钟消费者会被认为“死掉”触发rebalance。更尴尬的是rebalance期间消费组会停止消费积压会雪上加霜。所以如果你的单条消息处理比较慢适当降低max.poll.records或者提高max.poll.interval.ms都可以避免这种“假死”导致的循环rebalance。RocketMQ类似的参数是consumeThreadMin、consumeThreadMax以及ConsumeMessageBatchMaxSize。默认并发线程数是20如果积压严重可以先提高consumeThreadMax到40或50观察效果。但线程不是越多越好后面我会单独说。3.4 服务本身的瓶颈GC、CPU、数据库连接池消费线程本身没问题但整个服务资源被拖垮的情况也很常见。我遇到过一个问题消费者服务用的老机器堆内存设置太小消费高峰期频繁Full GC每次GC停顿好几秒。停顿期间消费线程完全不干活消息只能堆积。你能从监控上看到积压上涨的时间点和GC暂停的曲线几乎重合。数据库连接池用尽也是隐形杀手。消费者处理消息时通常要写数据库如果数据库连接池最大连接数是20而消费线程有30个其他线程只能排队等待连接消息处理时间就会直线上升。排查时可以监控连接池的等待时间和活跃连接数。还有一个容易被忽略的是日志同步写。消费速度快的时候如果每个消息都打印超长日志磁盘IO被打满线程也会卡在写日志上。我在优化高吞吐消费时都会把消费日志从DEBUG调整成WARN或者使用异步日志磁盘负载能明显降下来。4. 消费速度优化的实战手段4.1 调整并发与预取参数如果你已经定位到问题是消费能力不足而业务逻辑本身没有太大的优化空间那第一步就是调整消费者并发和预取参数。这里我直接给出可以“抄作业”的起步配置。RabbitMQ的Spring Boot配置以手动方式展示spring: rabbitmq: listener: simple: concurrency: 10 max-concurrency: 20 prefetch: 30 acknowledge-mode: manualconcurrency建议从CPU核数的2倍开始尝试。比如一台4核的机器先设8个消费者线程。prefetch可以结合单条处理时间如果单条消息平均处理100ms一个消费者每秒能处理10条那么prefetch设30相当于3秒的缓冲既不会让消费者空闲又不会导致消息全部堆积在Unacked。如果你的消费者处理逻辑较重建议prefetch设小一点。Kafka消费者的姿势不太一样。Kafka的并行度取决于分区数和消费者实例数。例如一个Topic有12个分区消费者组里有3个实例每个实例可以同时消费4个分区。但如果你只有一个消费者实例无论有多少分区它都得串行poll、串行处理。此时最优解是把poll到的消息放到一个自建线程池里去并发处理同时要控制好offset提交时机。我写过一个比较稳的模型while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); // 将records提交到线程池异步处理 executor.submit(() - processRecords(records)); // 等一批处理完后再手动提交offset consumer.commitSync(); }当然这只是最简版本实际还要处理线程池队列积压和停止时的优雅关闭。如果不想踩太多坑更推荐直接增加消费者实例数量让每个实例负责少数分区。4.2 批量消费与异步化改造单条消息单独消费如果消息量特别大每次都要建立一次数据库连接、发送一次网络请求性能一定上不去。批量消费往往是立竿见影的优化方式。RabbitMQ的Spring Boot可以通过SimpleRabbitListenerContainerFactory设置batchListenertrue然后在RabbitListener方法入参改为List 一次性处理一批消息。Kafka本身poll每次就能返回一批消息你要做的是在处理时批量入库、批量调用。RocketMQ也支持批量消费消息可以通过设置consumeMessageBatchMaxSize参数把一批消息一起交给消费者处理。批量消费最核心的收益是减少IO次数。举个例子以前100条消息要执行100次单条INSERT现在合成10次批量INSERT每次10条数据库写入耗时能降低一个数量级。前提是业务上允许一批消息一起处理并且不会产生过大的事务范围。异步化也是提高消费吞吐的利器。一种比较安全的做法是消费者线程收到消息后只做必要的校验和持久化然后把真正的业务处理丢给另一个独立的线程池立刻返回Ack。这样消费者的拉取速度不会受业务处理速度影响积压会快速下降。不过要注意这种方式可能会丢消息或重复消费必须保证后续处理线程任务的可靠性同时在系统重启时要能恢复未完成的任务。如果业务对最终一致性要求很高建议还是保持“处理完再提交offset/ack”的同步模式。4.3 横向扩容消费者实例与分区设计机器能加就加。横向扩容在MQ消费优化里是效果最直观、风险最低的手段。但扩容之前先想清楚你的消息模型支不支持扩容。Kafka的消费并行度强依赖分区数。同一个消费组里一个分区同时只能被一个消费者实例消费。如果你Topic只有6个分区消费者组里加再多实例最多也只有6个消费者在干活其他实例闲着。所以Kafka扩容前要先评估Topic是否需要增加分区数。增加分区会影响key对应的分区顺序和下游使用逻辑需要谨慎操作。一般建议在Topic创建初期就把分区设置成预估峰值并发数的两倍左右。RabbitMQ的队列天然支持多个消费者同时消费同一队列消费者之间是竞争关系所以直接增加RabbitListener实例就能提升吞吐。但要注意如果开了prefetch且设得很高消息会被少数消费者抢走扩容效果会打折扣。所以扩容时把prefetch调小让新消费者有机会拿到消息。RocketMQ每个Topic默认有多个读写队列一个消费组里的消费者会按队列数量进行负载均衡。如果你发现消费者实例很多但队列很少同样会有实例闲置。你可以用mqadmin updateTopic命令增加写队列或读队列但改变会触发Rebalance建议在低峰期操作。扩容消费者实例时我会优先选择同一个消费组下新增实例而不是另起一个独立消费组。同组实例会分摊消息而不同组都会各消费一份全量消息这相当于多了一次消费不但不能解积压反而会增加MQ压力。4.4 积压严重时的降级与快速消化策略如果积压量已经非常大靠常规优化仍然要很久才能消化这时候要学会“止损”。最常用的办法是把积压的消息快速转移到一个临时队列或备份Topic让主队列先恢复正常流量然后再慢慢处理备份队列里的历史积压。还有一种思路是“丢卒保车”在业务允许的情况下对非核心消息直接丢弃或者只记录不处理。比如日志分析消息、统计消息少处理几分钟影响不大但订单状态消息绝不能丢。我习惯根据消息的业务重要级别分类给不同队列设置不同的死信策略和最大积压阈值。除此之外可以临时调高消费者的并发数量同时关掉一些非必要的业务逻辑比如减少写日志、去掉无用的数据组装、关闭一些实时计算等让消费者优先把消息快速Ack掉。等积压恢复正常以后再把这些功能加回来。这种“瞬间获取流量吞吐”的手段虽然粗暴但是很有效适合应急。5. 常见问题与排查技巧实录5.1 我踩过的几个经典坑第一个坑是RabbitMQ的prefetch设置过大。当时有个消费者处理每条消息需要查一次数据库单条平均300ms。我为了追求吞吐把prefetch调到了500结果一上线500条消息全被第一个消费者拉走其他两个消费者空闲Unacked疯狂上涨后续消息全卡在这个消费者上。后来把prefetch调到10三个消费者立刻均衡了积压也明显下降。后来我养成了习惯只要是慢消费者prefetch绝对不超过单个消费者的处理速率乘以一个合理缓冲时间。第二个坑是Kafka的rebalance风暴。线上Kafka消费者里调用了外部接口接口出现抖动单个消息处理耗时接近3秒。max.poll.interval.ms还是默认5分钟按理说不应该出问题但因为我设置了max.poll.records500一次poll就拉回500条500条串行处理直接奔着10分钟去了消费者被认为超时触发了rebalance。rebalance期间所有消费者都停止消费积压雪上加霜。后来我把max.poll.records调成了50并在poll循环里判断处理耗时如果单批处理可能超过4分钟就提前提交offset情况立刻好转。第三个坑是RocketMQ的消费线程池满了。当时消费线程设置consumeThreadMin5consumeThreadMax20某天业务量上涨后很多线程阻塞在数据库写超时上线程池全部占满新的消息进来根本没有空闲线程处理积压不断增加。我以为单纯调大线程数就能解决结果发现到30时数据库连接池不够用了。最后同时调大了数据库连接池和RocketMQ消费线程数才真正把吞吐拉起来。5.2 消息积压排查问题速查表这里整理一个速查表遇到积压问题可以直接对着排查。现象可能原因验证方法解决手段积压持续上涨消费者CPU低消费者线程阻塞在外部调用jstack查看线程状态看是否有WAITING外部调用加超时异步化改造RabbitMQ Ready高Unacked低消费者拉取慢并发不够查看消费者连接数调整concurrency调大消费者并发调大prefetchRabbitMQ Ready低Unacked高消息已被消费者拿走但处理慢看Unacked曲线看处理耗时调小prefetch优化业务逻辑Kafka LAG高CURRENT-OFFSET不动消费者卡死或rebalance看消费者日志执行jstack检查处理逻辑调整max.poll参数Kafka频繁rebalancepoll处理耗时超过max.poll.interval.ms查看rebalance日志和单批处理耗时降低max.poll.records提高max.poll.interval.msRocketMQ ThreadPool满线程阻塞在IO或数据库查看消费线程池状态调大线程数优化数据库GC导致消费停顿Full GC频繁看GC日志和监控曲线调整堆大小减少大对象单分区LAG特别高分区数据倾斜查看各个分区LAG分布检查消息key设计重新分配分区5.3 一个百试百灵的排查步骤最后分享一个我自己的排查顺序按照这个顺序走大部分积压问题都能在30分钟内定位到根因。第一步先看监控大盘确认积压量、消费速率、生产速率三条曲线。第二步看消费者服务的CPU、内存、GC、线程池状态排除资源瓶颈。第三步打开消费日志统计单条消息处理耗时尤其是P99和最大值。如果P99很高就顺着耗时链路找慢SQL和外部调用。第四步检查MQ控制台/命令里的关键指标RabbitMQ看UnackedKafka看LAGRocketMQ看消费位点。第五步根据现象对照上面的速查表选择最可能的方向去验证。这套顺序我用了很久核心思路是“从宏观到微观先看资源再看代码先看指标再看日志”不要一上来就去翻业务代码否则可能排查半天最后发现只是线程数配少了。希望这篇内容对你有用也欢迎分享你遇到过的奇葩积压案例。