MQ(RabbitMQ)消息重复消费问题的全面解决方案

MQ消息重复消费是分布式系统中的常见问题，主要由网络问题、消费者故障、消息重试机制等引起。以下是针对RabbitMQ的完整解决方案体系：

一、消息生产端解决方案

1. 消息幂等设计

全局唯一消息ID：

MessageProperties props = MessagePropertiesBuilder.newInstance()
    .setMessageId(UUID.randomUUID().toString())
    .build();
Message message = new Message(body.getBytes(), props);
rabbitTemplate.send(exchange, routingKey, message);

业务键去重：

props.setHeader("biz_key", "order_123_create");

2. 生产者确认模式

// 开启确认模式
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        // 消息发送失败处理
    }
});

// 开启返回模式
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
    // 消息路由失败处理
});

二、消息存储端解决方案

1. 消息持久化

// 设置消息持久化
props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

// 队列持久化
@Bean
public Queue durableQueue() {
    return new Queue("order.queue", true); // true表示持久化
}

2. 死信队列配置

@Bean
public Queue mainQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", "dlx.exchange");
    args.put("x-dead-letter-routing-key", "dlx.routingKey");
    return new Queue("order.queue", true, false, false, args);
}

三、消费端核心解决方案

1. 消费幂等实现

数据库方案：

@Transactional
public void processOrder(OrderMessage message) {
    if (orderLogRepository.existsByMessageId(message.getMessageId())) {
        return;
    }
    
    Order order = createOrder(message);
    orderLogRepository.save(new OrderLog(message.getMessageId()));
}

Redis方案：

public void processMessage(OrderMessage message) {
    String key = "order:msg:" + message.getMessageId();
    if (redisTemplate.opsForValue().setIfAbsent(key, "1", 24, TimeUnit.HOURS)) {
        // 处理业务逻辑
    }
}

2. 手动ACK机制

@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage message, Channel channel, 
        @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    try {
        // 业务处理
        channel.basicAck(tag, false); // 手动确认
    } catch (Exception e) {
        channel.basicNack(tag, false, true); // 重新入队
    }
}

3. 消费限流

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setPrefetchCount(10); // 每次只获取10条消息
    return factory;
}

四、分布式锁方案

1. Redis分布式锁

public void processWithLock(OrderMessage message) {
    String lockKey = "order:lock:" + message.getOrderId();
    try {
        boolean locked = redisLock.lock(lockKey, 10, TimeUnit.SECONDS);
        if (locked) {
            // 处理业务
        }
    } finally {
        redisLock.unlock(lockKey);
    }
}

2. 数据库乐观锁

@Transactional
public void updateWithVersion(Order order) {
    int affected = orderMapper.update(
        "update orders set status = #{status}, version = version + 1 " +
        "where id = #{id} and version = #{version}", order);
    if (affected == 0) {
        throw new OptimisticLockException();
    }
}

五、消息轨迹方案

1. 全链路追踪

// 发送端
props.setHeader("trace_id", UUID.randomUUID().toString());

// 消费端
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage message, 
        @Header("trace_id") String traceId) {
    MDC.put("trace_id", traceId);
    // 处理业务
}

2. 状态机设计

public enum OrderStatus {
    INIT(1), PROCESSING(2), COMPLETED(3), FAILED(4);
    // 状态转换逻辑...
}

public void processOrder(Order order) {
    if (!order.getStatus().canTransitionTo(OrderStatus.PROCESSING)) {
        return; // 状态检查
    }
    // 处理业务
}

六、RabbitMQ特有方案

1. 消息去重插件

rabbitmq-plugins enable rabbitmq_message_deduplication

配置策略：

rabbitmqctl set_policy dedup-policy "^dedup-queue$" \
    '{"deduplication-mode":"all","deduplication-header":"message_id"}' \
    --apply-to queues

2. 消息TTL设置

// 设置消息过期时间
props.setExpiration("60000"); // 60秒

// 队列级别TTL
args.put("x-message-ttl", 60000);

3. 优先级队列

args.put("x-max-priority", 10); // 设置最大优先级
props.setPriority(5); // 设置消息优先级

七、架构级解决方案

1. 消息表+定时任务

CREATE TABLE consumed_messages (
    consumer_id VARCHAR(64),
    message_id VARCHAR(64) PRIMARY KEY,
    consumed_at TIMESTAMP,
    biz_key VARCHAR(128)
);

2. 去重服务设计

public class DeduplicationService {
    private final BloomFilter<String> bloomFilter;
    private final RedisTemplate<String, String> redisTemplate;
    
    public boolean isDuplicate(String messageId) {
        if (bloomFilter.mightContain(messageId)) {
            return redisTemplate.hasKey("msg:" + messageId);
        }
        return false;
    }
}

八、消费者配置优化

1. 消费者重试策略

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000

2. 死信队列处理

@RabbitListener(queues = "dlx.order.queue")
public void handleDeadLetter(OrderMessage message) {
    // 记录日志或人工干预
    alertService.notifyAdmin(message);
}

九、解决方案选择矩阵

十、最佳实践建议

1. 多级防御：生产端ID+消费端幂等+存储去重

2. 监控告警：监控重复消费率和死信队列

3. 压力测试：模拟网络分区和消费者重启

4. 日志完善：记录消息全生命周期日志

5. 版本兼容：消息体结构要向前兼容

RabbitMQ消息去重的关键在于根据业务场景选择合适的技术组合，通常建议：

• 简单场景：消息ID+Redis去重

• 金融场景：数据库唯一约束+事务

• 高并发场景：本地缓存+Redis校验

• 复杂业务：状态机+分布式锁

同时要注意消息堆积时的处理能力和去重存储的容量规划。