1. 消息重复消费问题的本质
在分布式消息系统中，消息重复消费是个经典难题。我经历过一个电商项目，促销期间由于消息重复导致同一个订单被处理了三次，差点造成重大损失。这个问题看似简单，实际涉及消息系统的多个核心机制。
消息重复的根本原因在于：消息系统无法保证"Exactly-Once"的语义。就像快递员送包裹，当网络抖动或超时发生时，发件人无法确定对方是否收到，只能选择重发。在RocketMQ中，这种"至少一次"的投递机制必然会导致重复消费的可能。
2. RocketMQ的消息投递机制解析
2.1 消息消费的完整生命周期
一个消息从生产到消费完成，会经历以下关键阶段：
- 生产者发送消息到Broker
- Broker持久化消息并返回ACK
- 消费者拉取消息
- 消费者处理完成后提交消费位点(offset)
- Broker标记消息为已消费
2.2 可能产生重复的关键环节
在实际运行中，以下场景会导致重复：
- 网络抖动：消费者处理成功但提交offset失败
- 客户端重启：内存中已处理但未持久化的offset丢失
- 负载均衡：消费者重新分配队列时offset回滚
- 事务消息：二阶段提交失败后的回查重试
3. 消息去重的核心解决方案
3.1 幂等性设计（推荐方案）
这是最根本的解决方案。就像银行转账系统，无论操作多少次，最终结果应该一致。实现方式包括：
// 使用Redis实现简易幂等控制 public boolean processMessage(Message message) { String msgId = message.getMsgId(); // SETNX + 过期时间 Boolean result = redisTemplate.opsForValue() .setIfAbsent("MSG_ID:" + msgId, "1", 24, TimeUnit.HOURS); if (!result) { return false; // 已处理过 } // 实际业务处理 return doRealBusiness(message); }关键参数选择：
- 过期时间：建议大于消息最大保留时间（默认3天）
- Key设计：建议包含业务标识如"ORDER_MSG:{msgId}"
3.2 事务状态表方案
适合强一致性要求的场景，如金融交易：
CREATE TABLE msg_transaction ( id BIGINT PRIMARY KEY, biz_id VARCHAR(64) UNIQUE, -- 业务唯一ID status TINYINT, -- 0未处理 1处理中 2处理成功 create_time DATETIME, update_time DATETIME, INDEX idx_biz_id (biz_id) );处理流程：
- 开启本地事务
- 插入或更新状态记录
- 执行业务逻辑
- 提交事务
3.3 RocketMQ自身机制
4.5+版本支持的消息去重功能：
# broker配置 enableConsumeMessageConcurrently=false enableConsumeMessageOrderly=true4. 不同场景下的方案选型
4.1 普通消息场景
| 方案 | 适用场景 | 优缺点 |
|---|---|---|
| 幂等设计 | 大部分业务场景 | 实现简单，但对代码有侵入性 |
| 状态表 | 强一致性要求 | 可靠但性能较低 |
| 消息属性 | 简单业务 | 依赖客户端实现 |
4.2 顺序消息场景
顺序消息需要特别注意：
- 必须保证同一个消息队列(MessageQueue)
- 消费失败会阻塞后续消息
- 建议配合本地事务实现
// 顺序消息消费示例 consumer.registerMessageListener(new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { try { // 业务处理 return ConsumeOrderlyStatus.SUCCESS; } catch (Exception e) { return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; } } });5. 实战中的经验技巧
5.1 消息轨迹追踪
建议为每条消息添加唯一traceId：
Message message = new Message(); message.putUserProperty("traceId", UUID.randomUUID().toString());5.2 消费监控配置
关键监控指标：
# 消费超时时间（默认15分钟） rocketmq.consumer.consumeTimeout=15m # 最大重试次数（默认16次） rocketmq.consumer.maxReconsumeTimes=165.3 常见问题排查
消息堆积导致重复：
- 检查消费者线程数：
consumeThreadMin/consumeThreadMax - 优化消费逻辑耗时
- 检查消费者线程数：
重复消费但无异常日志：
- 检查是否为网络分区导致
- 验证Broker的HA切换记录
事务消息重复：
-- 检查半事务消息 SELECT * FROM rmq_half_message WHERE msg_id = 'xxx';
6. 高级场景解决方案
6.1 分布式锁方案
对于跨系统调用的场景：
RedissonClient redisson = Redisson.create(); RLock lock = redisson.getLock("MSG_LOCK:" + message.getMsgId()); try { if (lock.tryLock(3, 30, TimeUnit.SECONDS)) { // 处理业务 } } finally { lock.unlock(); }6.2 消息延迟去重
适用于最终一致性场景：
// 使用延迟双删策略 redisTemplate.opsForValue().set(key, value, 30, TimeUnit.SECONDS); // 业务处理... // 30秒后异步删除 asyncTaskExecutor.execute(() -> { Thread.sleep(30000); redisTemplate.delete(key); });6.3 批处理消息去重
批量消费时的优化方案：
public void processBatch(List<Message> messages) { Map<String, Message> uniqueMessages = messages.stream() .collect(Collectors.toMap( Message::getMsgId, Function.identity(), (oldVal, newVal) -> oldVal)); // 处理去重后的消息 }7. 性能优化建议
Redis集群选择：
- 消息去重的Redis建议使用集群版
- 合理设置分片大小避免热点
本地缓存配合：
// Guava Cache做一级缓存 Cache<String, Boolean> localCache = CacheBuilder.newBuilder() .maximumSize(100000) .expireAfterWrite(1, TimeUnit.HOURS) .build();消息过滤优化：
// 使用SQL表达式过滤 consumer.subscribe("TopicTest", MessageSelector.bySql("TAGS IS NOT NULL AND created_time > 1698765432"));
在实际项目中，我们综合使用幂等设计+状态表的方式，将重复消费率从最初的0.3%降到了0.001%以下。关键是要根据业务特点选择合适的技术组合，并建立完善的监控体系。
