大宗商品销采运储系统・六层架构连载公共中间件层上一篇消息队列中篇我们重点拆解了事务消息和业务幂等解决销采运储单据场景下数据库事务与消息投递的原子性、消息重复消费脏数据两大核心问题。事务消息保障事件可靠发出幂等控制防止重复处理。但系统上线运行之后大量的故障不是来自业务代码而是消息队列运维层面消息堆积无限重试引发重试风暴异常消息污染正常队列集群无告警等到故障爆发才发现。本篇聚焦生产运维治理讲解消息队列集群的异常治理、监控告警、故障应急方案完成消息队列整套落地闭环。一句话定调事务消息和幂等解决业务正确性堆积治理、死信、监控告警解决系统稳定性。一、消息堆积定位原因与分级治理消息堆积是 RocketMQ 最常见的线上问题。本质一句话生产者生产速度 消费者消费速度。大宗商品业务极易出现堆积场景月底批量对账、夜间大批量单据同步、开盘瞬间大量订单涌入。1. 堆积常见根因消费逻辑过重消费内部包含复杂查询、大事务、同步调用外部接口单条消息处理耗时久下游服务故障数据库慢、第三方接口超时消费卡住消费实例不足并发线程数配置过小消费能力不足异常消息持续重试坏消息反复重试占用消费线程拖垮整个队列Topic 分区数不足分区决定消费最大并发分区太少无法横向扩容。2. 分级治理方案级别措施说明紧急预案临时扩容消费者实例提升消费并发快速消化积压消息短期优化拆分消费逻辑剥离非核心逻辑优化 SQL 与索引长期架构按业务域拆分 Topic大流量单据和普通通知消息隔离容量规划合理规划分区数量分区数决定消费并发上限工程规范禁止无限堆积设置堆积阈值告警不要等到消息堆积几十万条之后再处理。3. 堆积监控命令示例# 查看 Topic 堆积情况mqadmin consumerProgress -n 10.0.0.20:9876 -g stock-consumer-group# 查看指定 Topic 的消费 TPSmqadmin topicStatus -n 10.0.0.20:9876 -t supply_order_stock_topi二、重试风暴与死信队列完整治理1. 重试机制的坑RocketMQ 消费失败默认会自动重试。如果一条业务校验失败的消息反复重试会持续占用消费线程甚至把正常消息阻塞引发重试风暴。区分异常类型设置不同策略异常类型示例处理策略临时性异常网络抖动、数据库短暂超时有限次数重试指数退避拉长重试间隔业务非法异常单据不存在、参数错误、业务规则不满足禁止重试直接投递死信队列2. 重试配置示例rocketmq:consumer:# 最大重试次数max-reconsume-times: 3# 重试间隔毫秒指数退避suspend-current-queue-time-millis: 10003. 死信队列设计多次重试依然处理失败的消息自动转入死信 Topic。死信不是丢弃消息而是故障消息的隔离区核心能力独立 Topic和正常业务消息物理隔离坏消息不会污染主业务队列完整记录保存原始消息内容、重试次数、失败堆栈、发生时间管理后台查看消息、重发、人工补偿、丢弃。4. 死信队列配置示例RocketMQMessageListener( topic supply_order_stock_topic, consumerGroup stock-consumer-group, maxReconsumeTimes 3 ) public class StockOrderConsumer implements RocketMQListenerOrderStockDTO { Override public void onMessage(OrderStockDTO message) { try { stockService.createStockOrder(message); } catch (BusinessException e) { // 业务非法异常禁止重试直接转入死信 log.error(业务异常消息转入死信orderNo{}, message.getOrderNo(), e); throw new MessageBizException(e.getMessage()); } // 临时异常继续抛出RocketMQ 自动重试 } }销采运储业务落地规范所有核心单据 Topic 必须配置对应的死信 Topic死信消息产生后触发告警运维人员及时介入排查不能放任不管。三、集群监控与告警体系搭建只靠人工登录后台查看集群无法提前发现隐患需要一套完整监控指标体系。1. 核心监控指标层级指标说明集群Broker 节点状态、磁盘使用率、内存、连接数判断集群健康度Topic生产 TPS、消费 TPS、消息堆积数量判断 Topic 流量与积压消费组消费延迟、消费失败率、重试消息数、死信消息数判断消费健康度2. 告警分级级别触发条件通知方式P0 紧急Broker 节点宕机、磁盘满、大量死信产生电话 短信P1 重要消息堆积超过阈值、消费失败率突增企业微信/钉钉P2 观察TPS 波动、延迟小幅上涨日志记录每日报表汇总3. Prometheus 监控指标示例# RocketMQ Exporter 关键指标 rocketmq_producer_tps{topicsupply_order_stock_topic} rocketmq_consumer_tps{groupstock-consumer-group} rocketmq_group_diff{groupstock-consumer-group} # 消费堆积量 rocketmq_group_get_latency_by_storetime{groupstock-consumer-group} # 消费延迟 rocketmq_producer_offset{topicsupply_order_stock_topic} rocketmq_consumer_offset{groupstock-consumer-group}4. 告警规则示例groups:- name: rocketmq-alertrules:- alert: RocketMQMessageAccumulationexpr: rocketmq_group_diff{groupstock-consumer-group} 10000for: 5mlabels:severity: P1annotations:summary: 消费组 {{ $labels.group }} 消息堆积超过 1 万条- alert: RocketMQDeadLetterexpr: increase(rocketmq_group_dead_letter_count[5m]) 0labels:severity: P0annotations:summary: 消费组 {{ $labels.group }} 产生死信消息落地要求监控数据接入系统现有链路监控、日志服务把消息队列指标统一纳入大盘。四、生产环境故障应急案例月底对账消息堆积场景月底财务对账任务批量触发大量对账消息涌入 Topic消费处理慢消息持续堆积。应急步骤监控触发堆积告警收到通知排查消费日志确认无大量业务异常消息属于流量突增临时扩容消费者实例提升消费并发快速消化积压优化对账消费逻辑拆分大事务减少单条消息处理耗时事后复盘调整任务调度时间将批量对账任务错峰执行避开业务高峰。如果排查发现是非法业务消息导致持续重试则暂停消费把坏消息转入死信恢复正常队列消费。五、消息队列整套系列总结消息队列三篇连载到此完成篇核心内容解决什么问题上篇基础架构、流转模型、选型、基础代码理解消息队列能干什么中篇事务消息、业务幂等解决核心单据数据一致性下篇堆积治理、重试风暴、死信、监控告警保障生产环境长期稳定运行一句话总结整套消息队列消息队列通过异步解耦、削峰填谷提升系统弹性事务消息保障事件可靠投递幂等控制防止重复业务死信、监控、堆积治理守住生产稳定性底线共同支撑大宗商品销采运储全链路单据流转。下篇预告进入 Redis 缓存体系架构上篇讲解集群架构、多级缓存、热点 Key 治理以及大宗商品库存、商品报价、客户基础信息的缓存实战落地。
大宗商品销采运储系统・六层架构连载公共中间件层上一篇消息队列中篇我们重点拆解了事务消息和业务幂等解决销采运储单据场景下数据库事务与消息投递的原子性、消息重复消费脏数据两大核心问题。事务消息保障事件可靠发出幂等控制防止重复处理。但系统上线运行之后大量的故障不是来自业务代码而是消息队列运维层面消息堆积无限重试引发重试风暴异常消息污染正常队列集群无告警等到故障爆发才发现。本篇聚焦生产运维治理讲解消息队列集群的异常治理、监控告警、故障应急方案完成消息队列整套落地闭环。一句话定调事务消息和幂等解决业务正确性堆积治理、死信、监控告警解决系统稳定性。一、消息堆积定位原因与分级治理消息堆积是 RocketMQ 最常见的线上问题。本质一句话生产者生产速度 消费者消费速度。大宗商品业务极易出现堆积场景月底批量对账、夜间大批量单据同步、开盘瞬间大量订单涌入。1. 堆积常见根因消费逻辑过重消费内部包含复杂查询、大事务、同步调用外部接口单条消息处理耗时久下游服务故障数据库慢、第三方接口超时消费卡住消费实例不足并发线程数配置过小消费能力不足异常消息持续重试坏消息反复重试占用消费线程拖垮整个队列Topic 分区数不足分区决定消费最大并发分区太少无法横向扩容。2. 分级治理方案级别措施说明紧急预案临时扩容消费者实例提升消费并发快速消化积压消息短期优化拆分消费逻辑剥离非核心逻辑优化 SQL 与索引长期架构按业务域拆分 Topic大流量单据和普通通知消息隔离容量规划合理规划分区数量分区数决定消费并发上限工程规范禁止无限堆积设置堆积阈值告警不要等到消息堆积几十万条之后再处理。3. 堆积监控命令示例# 查看 Topic 堆积情况mqadmin consumerProgress -n 10.0.0.20:9876 -g stock-consumer-group# 查看指定 Topic 的消费 TPSmqadmin topicStatus -n 10.0.0.20:9876 -t supply_order_stock_topi二、重试风暴与死信队列完整治理1. 重试机制的坑RocketMQ 消费失败默认会自动重试。如果一条业务校验失败的消息反复重试会持续占用消费线程甚至把正常消息阻塞引发重试风暴。区分异常类型设置不同策略异常类型示例处理策略临时性异常网络抖动、数据库短暂超时有限次数重试指数退避拉长重试间隔业务非法异常单据不存在、参数错误、业务规则不满足禁止重试直接投递死信队列2. 重试配置示例rocketmq:consumer:# 最大重试次数max-reconsume-times: 3# 重试间隔毫秒指数退避suspend-current-queue-time-millis: 10003. 死信队列设计多次重试依然处理失败的消息自动转入死信 Topic。死信不是丢弃消息而是故障消息的隔离区核心能力独立 Topic和正常业务消息物理隔离坏消息不会污染主业务队列完整记录保存原始消息内容、重试次数、失败堆栈、发生时间管理后台查看消息、重发、人工补偿、丢弃。4. 死信队列配置示例RocketMQMessageListener( topic supply_order_stock_topic, consumerGroup stock-consumer-group, maxReconsumeTimes 3 ) public class StockOrderConsumer implements RocketMQListenerOrderStockDTO { Override public void onMessage(OrderStockDTO message) { try { stockService.createStockOrder(message); } catch (BusinessException e) { // 业务非法异常禁止重试直接转入死信 log.error(业务异常消息转入死信orderNo{}, message.getOrderNo(), e); throw new MessageBizException(e.getMessage()); } // 临时异常继续抛出RocketMQ 自动重试 } }销采运储业务落地规范所有核心单据 Topic 必须配置对应的死信 Topic死信消息产生后触发告警运维人员及时介入排查不能放任不管。三、集群监控与告警体系搭建只靠人工登录后台查看集群无法提前发现隐患需要一套完整监控指标体系。1. 核心监控指标层级指标说明集群Broker 节点状态、磁盘使用率、内存、连接数判断集群健康度Topic生产 TPS、消费 TPS、消息堆积数量判断 Topic 流量与积压消费组消费延迟、消费失败率、重试消息数、死信消息数判断消费健康度2. 告警分级级别触发条件通知方式P0 紧急Broker 节点宕机、磁盘满、大量死信产生电话 短信P1 重要消息堆积超过阈值、消费失败率突增企业微信/钉钉P2 观察TPS 波动、延迟小幅上涨日志记录每日报表汇总3. Prometheus 监控指标示例# RocketMQ Exporter 关键指标 rocketmq_producer_tps{topicsupply_order_stock_topic} rocketmq_consumer_tps{groupstock-consumer-group} rocketmq_group_diff{groupstock-consumer-group} # 消费堆积量 rocketmq_group_get_latency_by_storetime{groupstock-consumer-group} # 消费延迟 rocketmq_producer_offset{topicsupply_order_stock_topic} rocketmq_consumer_offset{groupstock-consumer-group}4. 告警规则示例groups:- name: rocketmq-alertrules:- alert: RocketMQMessageAccumulationexpr: rocketmq_group_diff{groupstock-consumer-group} 10000for: 5mlabels:severity: P1annotations:summary: 消费组 {{ $labels.group }} 消息堆积超过 1 万条- alert: RocketMQDeadLetterexpr: increase(rocketmq_group_dead_letter_count[5m]) 0labels:severity: P0annotations:summary: 消费组 {{ $labels.group }} 产生死信消息落地要求监控数据接入系统现有链路监控、日志服务把消息队列指标统一纳入大盘。四、生产环境故障应急案例月底对账消息堆积场景月底财务对账任务批量触发大量对账消息涌入 Topic消费处理慢消息持续堆积。应急步骤监控触发堆积告警收到通知排查消费日志确认无大量业务异常消息属于流量突增临时扩容消费者实例提升消费并发快速消化积压优化对账消费逻辑拆分大事务减少单条消息处理耗时事后复盘调整任务调度时间将批量对账任务错峰执行避开业务高峰。如果排查发现是非法业务消息导致持续重试则暂停消费把坏消息转入死信恢复正常队列消费。五、消息队列整套系列总结消息队列三篇连载到此完成篇核心内容解决什么问题上篇基础架构、流转模型、选型、基础代码理解消息队列能干什么中篇事务消息、业务幂等解决核心单据数据一致性下篇堆积治理、重试风暴、死信、监控告警保障生产环境长期稳定运行一句话总结整套消息队列消息队列通过异步解耦、削峰填谷提升系统弹性事务消息保障事件可靠投递幂等控制防止重复业务死信、监控、堆积治理守住生产稳定性底线共同支撑大宗商品销采运储全链路单据流转。下篇预告进入 Redis 缓存体系架构上篇讲解集群架构、多级缓存、热点 Key 治理以及大宗商品库存、商品报价、客户基础信息的缓存实战落地。
