当前位置：首页 > java >正文

RabbitMQ消息堆积问题排查：concurrentConsumers 配置的坑与解决方案

java 2025/9/3 9:01:05

一、问题背景

在分布式系统中，消息队列常用于异步处理和解耦。某业务场景下，用户完成支付后，支付结果通过MQ通知业务系统更新状态。但用户反馈支付成功后，需要等待好几分钟才能在业务系统中看到状态更新，体验较差。

二、问题现象

业务流程如下：

用户支付成功
支付渠道将支付结果发送到MQ队列
业务服务消费MQ中的支付结果
消费成功后更新业务单据状态
向相关系统发送支付结果通知

监控发现，从消息生产到消费完成，平均耗时约2分钟，无法满足业务实时性要求。

三、问题排查

1. 队列堆积检查

使用RabbitMQ管理命令检查队列状态：
rabbitmqctl list_queues name messages messages_unacknowledged

监控数据显示：

16:54:00 queue_async_notify      2
....
16:59:30 queue_async_notify      0

明显看到消息在队列中堆积，直到16:58:30才被消费完。

2. 消费线程分析

检查业务日志发现，所有消息都由同一个线程处理：

3. 代码配置检查

查看RabbitMQ配置类发现关键问题：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter msgConverter) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(msgConverter);factory.setMaxConcurrentConsumers(10); // 只设置了最大并发数return factory;
}

4. 根本原因分析

• 业务处理平均耗时20秒，属于IO密集型操作

• 配置中只设置了maxConcurrentConsumers，未设置concurrentConsumers

• 由于消息量不大，未触发动态扩容机制

• 实际并发消费者数保持默认值1，导致消息串行处理

• 单线程处理速度跟不上消息生产速度，造成堆积

四、解决方案

方案一：固定并发消费者数（推荐）

@Bean
public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(
ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

// 设置固定并发消费者数
factory.setConcurrentConsumers(5);
return factory;

}

方案二：动态扩容配置

// 动态扩容配置
factory.setConcurrentConsumers(1);           // 初始并发数
factory.setMaxConcurrentConsumers(5);        // 最大并发数
factory.setPrefetchCount(1);                // 预取消息数
factory.setConsecutiveActiveTrigger(3);      // 连续活跃触发扩容
factory.setConsecutiveIdleTrigger(3);       // 连续空闲触发缩容
factory.setReceiveTimeout(1000L);            // 接收超时时间
return factory;

}