1. 问题背景：消息为何会重复投递？

在咱们日常使用RabbitMQ时，最常遇到的"幽灵问题"就是消息重复消费。这就像网购时点击支付按钮后网络卡顿，结果重复提交了订单一样让人头疼。RabbitMQ本身提供的"至少一次"投递保证机制，在生产者确认、消费者确认等环节都可能产生重复：

  • 生产者重试机制：当网络抖动时，消息可能已成功发送但没收到Broker确认
  • 消费者确认丢失：消费者处理完消息但确认消息时连接断开
  • 队列重新入队：消费者处理超时导致消息重新回到队列

举个实际案例：某电商系统每天有3%的订单重复支付，经排查发现是支付回调消息重复触发导致的。这就是典型的需要消息去重的场景。

2. 解决方案三部曲

2.1 幂等性设计（业务层防护）

// 使用Dapper作为ORM示例
public class OrderService
{
    // 创建订单的幂等方法
    public bool CreateOrder(string orderId, decimal amount)
    {
        using (var conn = new SqlConnection(_connectionString))
        {
            // 先检查订单是否已存在（关键步骤）
            var existing = conn.QueryFirstOrDefault<Order>(
                "SELECT * FROM Orders WHERE OrderId = @orderId", 
                new { orderId });
                
            if (existing != null) return true; // 已存在直接返回
            
            // 插入新订单（数据库唯一索引兜底）
            var result = conn.Execute(
                "INSERT INTO Orders(OrderId, Amount, Status) VALUES(@orderId, @amount, 'created')",
                new { orderId, amount });
                
            return result > 0;
        }
    }
}

技术要点：

  • 数据库建立OrderId唯一索引作为最终防线
  • 查询先行降低插入冲突概率
  • 适用于写操作类的业务场景

2.2 消息去重表（中间件层防护）

// 使用Redis作为去重存储
public class MessageDeduplicator
{
    private readonly IDatabase _redis;
    
    public MessageDeduplicator(IConnectionMultiplexer redis)
    {
        _redis = redis.GetDatabase();
    }

    public bool IsProcessed(string messageId)
    {
        // 使用Redis的原子操作保证并发安全
        return _redis.StringSet($"msg:{messageId}", "1", 
            expiry: TimeSpan.FromHours(24), 
            when: When.NotExists);
    }
}

// 消费者应用示例
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) => {
    var messageId = ea.BasicProperties.MessageId;
    
    if (!deduplicator.IsProcessed(messageId))
    {
        // 处理业务逻辑...
        channel.BasicAck(ea.DeliveryTag, false);
    }
    else
    {
        channel.BasicReject(ea.DeliveryTag, false);
    }
};

实现技巧：

  • Redis SETNX命令实现原子操作
  • 消息ID建议使用业务标识+时间戳组合
  • 设置合理过期时间（建议大于最大可能处理时间）

2.3 版本号控制（数据层防护）

public class InventoryMessage
{
    public string ProductId { get; set; }
    public int Quantity { get; set; }
    public int Version { get; set; } // 版本号字段
}

// 处理库存更新
public void HandleInventoryUpdate(InventoryMessage message)
{
    using (var transaction = new TransactionScope())
    {
        var current = GetProductVersion(message.ProductId);
        if (message.Version <= current) return;
        
        UpdateInventory(message.ProductId, message.Quantity);
        UpdateProductVersion(message.ProductId, message.Version);
        
        transaction.Complete();
    }
}

设计要点：

  • 版本号建议使用时间戳或序列号
  • 数据库事务保证版本检查和更新的原子性
  • 适用于需要顺序处理的场景

3. 应用场景分析

3.1 电商交易系统

  • 支付回调：使用Redis去重表+订单号校验
  • 库存扣减：版本号控制防止超卖
  • 订单状态流转：数据库幂等设计

3.2 物联网数据采集

  • 设备状态上报：消息TTL+版本号控制
  • 批量数据处理：BloomFilter预处理

3.3 金融交易系统

  • 账户余额变更：数据库事务+版本号
  • 对账文件生成：Redis原子计数器

4. 技术方案对比

方案类型 适用场景 优点 缺点
业务幂等设计 写操作类业务 实现简单，业务强一致 依赖数据库性能
消息去重表 高频消息处理 处理速度快，扩展性好 需要额外存储
版本号控制 顺序敏感型操作 天然防重复，支持顺序性 增加消息设计复杂度

5. 避坑指南

5.1 防重复设计四大原则

  1. 唯一标识：消息必须携带全局唯一ID
  2. 原子操作：检查与处理要保证原子性
  3. 过期机制：存储数据要设置合理有效期
  4. 降级预案：去重服务故障时的应对策略

5.2 常见误区警示

  • 不要依赖RabbitMQ的自动重试机制
  • 避免在消费者内存中维护处理状态
  • 分布式锁不要滥用（建议使用CAS方式）
  • 消息ID不要使用自增ID（分库分表会重复）

6. 方案选型建议

根据我们的实战经验，给出以下决策树：

  1. 是否需要严格顺序 → 是 → 版本号方案
  2. 消息量是否巨大 → 是 → Redis去重表
  3. 是否已有业务状态 → 是 → 幂等设计
  4. 其他情况 → 组合使用两种方案

7. 总结与展望

解决消息重复问题就像给系统穿上一件防弹衣，需要根据业务特性选择合适的防护策略。随着业务规模的扩大，可以考虑以下优化方向：

  1. 混合方案：Redis+数据库的多级缓存设计
  2. 压缩优化：对消息ID进行编码压缩
  3. 监控体系：建立重复消息告警机制
  4. 新特性探索：RabbitMQ的仲裁队列特性

记住，没有完美的解决方案，只有最适合当前业务场景的方案。建议定期进行消息轨迹分析，持续优化去重策略，让消息队列真正成为系统可靠性的守护者而不是故障源。