1. 消息队列重复消费问题全景解析消息队列作为分布式系统解耦的利器在电商秒杀、日志收集等场景中广泛应用。但我在实际工作中发现重复消费问题如同附骨之疽——某次大促活动因重复发放优惠券导致百万损失让我们不得不重新审视这个经典问题。本质上重复消费源于消息队列的至少一次投递机制。当消费者处理完消息但未及时提交offset时系统重试或重启会导致消息被二次处理。这与TCP协议的重传机制类似都是通过冗余来保证可靠性但需要业务层自己处理重复。2. 消息幂等性设计实战2.1 数据库唯一索引方案在订单创建场景中我们为每个消息分配唯一业务ID如订单号在数据库建立唯一索引ALTER TABLE orders ADD UNIQUE INDEX uk_order_id (order_id);当重复消息触发重复插入时数据库会抛出DuplicateKeyException。但要注意高并发下建议配合SELECT FOR UPDATE先查后插避免唯一索引冲突导致事务回滚影响性能2.2 Redis原子操作方案利用Redis的SETNX指令实现分布式锁Boolean lock redisTemplate.opsForValue().setIfAbsent( order:lock: orderId, 1, 10, TimeUnit.SECONDS); if (!lock) { throw new RuntimeException(操作正在处理中); }实测中我们遇到锁过期但业务未完成的情况最终采用Redisson的看门狗机制自动续期。2.3 状态机版本控制对于复杂业务流程如订单状态流转我们在数据库增加version字段UPDATE order SET status PAID, version version 1 WHERE order_id ? AND version ?;版本号不匹配时影响行数为0可有效防止状态重复更新。3. 消息队列服务端配置优化3.1 Kafka精准一次语义通过配置enable.idempotencetrue开启生产者幂等props.put(enable.idempotence, true); props.put(acks, all);配合事务API实现跨分区原子写入producer.beginTransaction(); try { producer.send(record1); producer.send(record2); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); }3.2 RocketMQ消息重试策略在消费者端配置最大重试次数consumer.setMaxReconsumeTimes(3);我们曾因未设置重试上限导致死循环最终采用重试队列死信队列分级处理前3次快速重试后续每次间隔10秒16次后转入死信队列人工处理4. 消费者端防重复架构设计4.1 本地消息表方案在业务数据库创建消息日志表CREATE TABLE message_log ( msg_id VARCHAR(64) PRIMARY KEY, status TINYINT DEFAULT 0, created_time DATETIME );处理流程开启事务插入消息记录执行业务操作更新消息状态提交事务4.2 分布式事务方案采用Seata的AT模式GlobalTransactional public void handleMessage(OrderMessage message) { orderService.create(message); inventoryService.deduct(message); }关键配置seata.tx-service-groupmy_tx_group seata.service.vgroup-mapping.my_tx_groupdefault5. 典型场景解决方案实录5.1 支付回调处理支付平台可能重复推送通知我们采用前置校验检查商户订单号是否存在中间处理通过SELECT FOR UPDATE加锁后置确认异步发送MQ确认消息5.2 库存扣减场景采用CAS乐观锁方案UPDATE inventory SET stock stock - 1 WHERE item_id ? AND stock 1;配合Redis Lua脚本实现预扣减local key KEYS[1] local change tonumber(ARGV[1]) local current tonumber(redis.call(GET, key)) if current change then return redis.call(INCRBY, key, -change) end return -16. 监控与治理实践6.1 重复消息监控通过埋点统计重复率Around(annotation(idempotent)) public Object checkRepeat(ProceedingJoinPoint joinPoint) { String key buildKey(joinPoint); if (cache.contains(key)) { metrics.counter(repeat.msg).increment(); return null; } cache.put(key, ); return joinPoint.proceed(); }6.2 自动化治理方案搭建消息治理平台包含实时仪表盘展示各业务线重复率自动拦截对高频重复消息降级处理智能告警基于历史基线动态阈值告警在每天千万级消息量的系统中这套方案将重复消费率从0.3%降至0.002%。核心经验是没有银弹需要根据业务特征组合多种方案。比如支付业务采用DB唯一索引本地消息表而日志处理则用Redis指纹自动过期的轻量级方案。