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

记一次生产环境MQ队列积压150W问题分析与解决方案

news 2026/5/13 0:34:27

# MQ队列积压150W+问题分析与解决方案报告 ## 一、背景描述 ### 1.1 问题现象 - **队列积压量**：150W+ 消息 - **影响范围**：消息消费延迟严重，队列持续增长 - **风险等级**：🔴 **高危** - 存在MQ服务器内存溢出及宕机风险 ### 1.2 根因分析

┌─────────────────────────────────────────────────────────┐
│ 问题架构图示 │
├─────────────────────────────────────────────────────────┤
│ Producer ──► [MQ队列: 150W+积压] ◄── Consumer │
│ (生产者) 消息无差别投递 (消费者端过滤) │
│ │
│ ❌ 问题：过滤逻辑后置，导致无效消息大量堆积 │
│ │
│ 消费者处理流程： │
│ 接收消息 → 计算MD5 → 查重判断 → 重复则丢弃 │
│ ↑___________________________________________↓ │
│ (高CPU消耗操作在消费端执行) │
└─────────────────────────────────────────────────────────┘

| 维度 | 现状问题 | 理想状态 | |:---|:---|:---| | **过滤位置** | 消费者端执行 | 生产者端执行 | | **资源消耗** | 150W+次MD5计算 | 0次无效消息投递 | | **队列压力** | 无效消息占用存储 | 仅有效消息入队 | | **消费延迟** | 严重延迟 | 实时处理 | --- ## 二、MQ管理端操作简介 ### 2.1 常用管理工具 | 工具 | 访问方式 | 核心功能 | |:---|:---|:---| | **RabbitMQ Management** | `http://host:15672` | 可视化监控、队列管理 | | **RocketMQ Console** | 部署Web控制台 | Topic/ConsumerGroup管理 | | **Kafka UI (Kowl/AKHQ)** | 独立部署 | 分区监控、消息查询 | ### 2.2 关键监控指标 ```bash # RabbitMQ 命令行查看队列深度 rabbitmqctl list_queues name messages_ready messages_unacknowledged # 输出示例 # name messages_ready messages_unacknowledged # task_queue 1523421 0

2.3 积压应急操作（⚠️ 谨慎执行）

操作	命令/路径	适用场景
查看队列状态	Queues → 队列名 → Get messages	诊断消息内容
Purge清空队列	Queues → 队列名 → Purge	本次采用
Delete删除队列	Queues → 队列名 → Delete	重建队列
消费速率监控	Overview → Message rates	评估处理能力

Purge操作截图示意：

RabbitMQ Management → Queues → [队列名] → [Purge Messages] 按钮 → 确认"Are you sure?"

三、解决方案

3.1 临时方案：Purge清空队列

执行步骤

# 1. 确认积压队列名称rabbitmqctl list_queues|greptask# 2. 评估影响（可选：备份部分消息）# 通过管理界面导出或消费端采样# 3. 执行Purge（管理界面或API）curl-uuser:pass-XDELETE http://mq-host:15672/api/queues/%2f/task_queue/contents# 4. 验证清理结果rabbitmqctl list_queues name messages_ready

风险控制

风险点	应对措施
误删有效消息	业务低峰期执行；提前通知业务方
消息丢失不可恢复	明确接受临时方案的数据损失
消费者空转	临时降低消费者实例数

3.2 长久方案：前置过滤逻辑

架构改造

┌─────────────────────────────────────────────────────────┐ │ 优化后架构图示 │ ├─────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ 任务生产 │───►│ MD5查重服务 │───►│ MQ队列 │ │ │ │ (Producer) │ │ (新增) │ │ (精简有效) │ │ │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ │ ↑______________________________│ │ │ │ │ 重复任务直接过滤 │ ▼ │ │ └──────────────────────────────┘ ┌─────────────┐│ │ │ Consumer ││ │ │ (纯业务处理) ││ │ └─────────────┘│ │ │ │ ✅ 收益：MD5计算前置，无效消息0入队，队列压力降低90%+ │ └─────────────────────────────────────────────────────────┘

代码改造示例

// ========== 改造前：消费者端过滤（问题代码） ==========@ComponentpublicclassTaskConsumer{@RabbitListener(queues="task_queue")publicvoidconsume(Messagemessage){StringfilePath=parseMessage(message);// ❌ 问题：高耗操作在消费端，无效消息已占用队列Stringmd5=calculateMd5(filePath);// 150W+次执行if(md5Cache.exists(md5)){log.warn("重复文件，丢弃: {}",filePath);return;// 消息已投递，资源已浪费}processBusiness(filePath);// 实际业务}}// ========== 改造后：生产者端过滤 ==========@ServicepublicclassTaskProducer{@AutowiredprivateMd5Servicemd5Service;@AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidproduceTask(StringfilePath){// ✅ 优化：入队前完成过滤Stringmd5=md5Service.calculateMd5(filePath);if(md5Service.isDuplicate(md5)){log.info("重复文件，跳过投递: {}",filePath);return;// 直接过滤，不占用MQ资源}// 仅有效消息入队TaskMessagemsg=newTaskMessage(filePath,md5);rabbitTemplate.convertAndSend("task_exchange","task_routing",msg);}}@ComponentpublicclassOptimizedConsumer{@RabbitListener(queues="task_queue")publicvoidconsume(TaskMessagemessage){// ✅ 消费端专注业务，无需重复计算MD5processBusiness(message.getFilePath());}}

配套优化：MD5查重服务

@ServicepublicclassMd5Service{// 方案1：Redis Set（推荐，O(1)查询）@AutowiredprivateStringRedisTemplateredisTemplate;publicbooleanisDuplicate(Stringmd5){Booleanadded=redisTemplate.opsForSet().add("md5:set",md5);return!Boolean.TRUE.equals(added);// 已存在返回true}// 方案2：BloomFilter（超大规模，允许微量误判）@AutowiredprivateRBloomFilter<String>bloomFilter;publicbooleanmightDuplicate(Stringmd5){if(!bloomFilter.contains(md5)){bloomFilter.add(md5);returnfalse;// 一定不重复}returntrue;// 可能重复，需二次确认}}

四、方案对比与收益

指标	改造前	临时方案	长久方案
队列积压	150W+持续增长	清零	维持低位
MD5计算次数	150W+/批次	-	有效任务数
MQ存储压力	极高	缓解	极低
消费延迟	小时级	恢复	秒级
数据一致性	最终一致	可能丢失	最终一致
实施成本	-	5分钟	2-3天开发

五、实施时间线

Day 0 ──┬── 问题发现，队列积压150W+ │ Day 0 ──┼── [紧急] 执行Purge临时方案 ✅ 14:00 │ │ Day 1-2 ──┼── 开发长久方案：MD5查重服务 │ Day 3 ──┼── 联调测试，灰度发布 │ Day 4 ──┴── 全量上线，监控验证 ✅

六、经验总结

6.1 设计原则

“过滤前置，计算后置”— 昂贵操作尽量靠近数据源

6.2 监控建议

// 增加生产者端指标监控MeterRegistryregistry=...;registry.counter("mq.produce.filtered","reason","duplicate").increment();registry.counter("mq.produce.success").increment();