# 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 02.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管理界面或APIcurl-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(queuestask_queue)publicvoidconsume(Messagemessage){StringfilePathparseMessage(message);// ❌ 问题高耗操作在消费端无效消息已占用队列Stringmd5calculateMd5(filePath);// 150W次执行if(md5Cache.exists(md5)){log.warn(重复文件丢弃: {},filePath);return;// 消息已投递资源已浪费}processBusiness(filePath);// 实际业务}}// 改造后生产者端过滤 ServicepublicclassTaskProducer{AutowiredprivateMd5Servicemd5Service;AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidproduceTask(StringfilePath){// ✅ 优化入队前完成过滤Stringmd5md5Service.calculateMd5(filePath);if(md5Service.isDuplicate(md5)){log.info(重复文件跳过投递: {},filePath);return;// 直接过滤不占用MQ资源}// 仅有效消息入队TaskMessagemsgnewTaskMessage(filePath,md5);rabbitTemplate.convertAndSend(task_exchange,task_routing,msg);}}ComponentpublicclassOptimizedConsumer{RabbitListener(queuestask_queue)publicvoidconsume(TaskMessagemessage){// ✅ 消费端专注业务无需重复计算MD5processBusiness(message.getFilePath());}}配套优化MD5查重服务ServicepublicclassMd5Service{// 方案1Redis Set推荐O(1)查询AutowiredprivateStringRedisTemplateredisTemplate;publicbooleanisDuplicate(Stringmd5){BooleanaddedredisTemplate.opsForSet().add(md5:set,md5);return!Boolean.TRUE.equals(added);// 已存在返回true}// 方案2BloomFilter超大规模允许微量误判AutowiredprivateRBloomFilterStringbloomFilter;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();6.3 防护机制生产者端限流 重复校验MQ层队列长度告警阈值10W消费者端消费速率监控 自动扩容