当前位置：首页 > news >正文

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

news 来源：原创 2025/5/28 4:31:59

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: truemax-attempts: 3initial-interval: 1000

2. 死信队列处理

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

九、解决方案选择矩阵

场景	推荐方案	适用条件
金融交易	数据库唯一约束+事务	强一致性要求
高并发订单	Redis原子操作+本地缓存	高性能要求
长时间业务处理	手动ACK+分布式锁	处理耗时较长
全链路追踪	消息轨迹+状态机	复杂业务流程
海量消息	布隆过滤器+持久化存储	内存资源有限