在分布式消息队列中，消息重复消费是一个常见问题（例如网络重试、消费者重启等场景）。RocketMQ 通过多种机制来尽量减少重复消费，但无法完全避免（需业务方配合实现幂等性）。以下是 RocketMQ 的解决方案和最佳实践：

1. RocketMQ 自身的防重复机制

(1) 消息重试机制（Consumer端）

• 默认重试策略：若消费者消费失败（返回 RECONSUME_LATER），消息会被重新投递（最多 16 次）。
• 风险点：重试可能导致同一条消息被多次消费。

(2) 消息幂等性标识（Message ID + Offset）

• 每条消息有唯一的 Message ID 和 Broker Offset，消费者可通过记录已处理的 ID/Offset 去重。
• 局限性：极端情况下（如 Broker 主从切换），Message ID 可能重复（需结合业务键去重）。

2. 业务层必须实现的幂等性方案

RocketMQ 不保证绝对不重复，需业务代码自行处理幂等。常见方案：

(1) 唯一业务键 + 数据库去重

• 适用场景：订单ID、支付流水号等有唯一键的业务。
• 实现方式：
• java

// 伪代码：消费前检查唯一键是否已处理
String orderId = message.getUserProperty("order_id");
if (orderId != null && !isOrderProcessed(orderId)) {
    processOrder(orderId);
    markOrderAsProcessed(orderId); // 写入DB或缓存
}

• 存储选择：
• 数据库唯一索引：插入前检查 order_id 是否已存在。
• Redis SETNX：利用原子操作记录已处理的消息键。

(2) 乐观锁（适用于更新操作）

• 示例（库存扣减）：
• sql

UPDATE inventory SET stock = stock - 1 
WHERE product_id = #{productId} AND stock >= 1;

• 即使消息重复，数据库乐观锁会保证只扣减一次。

(3) 分布式锁（强一致性场景）

• 消费前先获取锁（如 Redis 的 SET key value NX EX）：
• java
• 
  

String lockKey = "msg_lock:" + message.getMsgId();
try {
    if (redisClient.setIfAbsent(lockKey, "1", 30, TimeUnit.SECONDS)) {
        processMessage(message);
    }
} finally {
    redisClient.delete(lockKey);
}

3. RocketMQ 4.x 后的增强特性

(1) 事务消息

• 二阶段提交机制可减少因生产者重试导致的消息重复（但消费者仍需幂等）。
• 适用场景：跨系统事务（如支付后发消息）。

(2) 消息轨迹（Message Trace）

• 记录消息生命周期（发送、存储、消费），便于排查重复消费问题。

4. 最佳实践总结

代码示例（消费者幂等）

java

// 基于 Redis 的幂等消费示例
public class MyConsumer implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        for (MessageExt message : messages) {
            String msgId = message.getMsgId();
            String orderId = message.getUserProperty("order_id");


            // 1. 检查 Redis 是否已处理
            if (redisTemplate.opsForValue().setIfAbsent("consumed:" + orderId, "1", 24, TimeUnit.HOURS)) {
                try {
                    // 2. 业务处理
                    processOrder(orderId);
                } catch (Exception e) {
                    redisTemplate.delete("consumed:" + orderId); // 失败时清除标记
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            } else {
                System.out.println("消息已处理，跳过: " + orderId);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

关键结论

• RocketMQ 不保证 100% 不重复：需业务方结合消息唯一标识（如订单ID）实现幂等。
• 轻量级方案：优先用 Redis/Database 去重；高并发场景可用乐观锁或分布式锁。
• 监控：通过日志或 RocketMQ Admin Tool 定期检查重复消息。