文章目录
消息消费中的可靠性问题
当我们处理消息时，尤其是异步消息队列系统（如RabbitMQ），常常面临网络或系统故障导致消息未确认或重复消费的问题。以下是常见的场景：
- 消费者成功处理业务，但网络故障导致未发送确认（
ack）消息，RabbitMQ认为消费者崩溃，将重新投递消息。 - 消费者已经处理过消息，然而消息再次投递时，导致重复处理相同的消息，从而引发业务逻辑问题（如库存重复扣减、余额重复扣款等）。
业务幂等性
- 幂等性定义：在编程中，幂等性意味着同一个操作执行一次或多次，结果一致，不会引发额外的副作用。具体而言，对于同一业务，不管是被执行一次还是多次，业务状态的影响应当一致。
- 幂等性的重要性：如果业务不具备幂等性，重复消费消息可能会引发严重的业务问题。例如，重复扣减库存或重复扣款。
消息重复消费的原因
- 消息确认机制中，如果消费者处理成功，但因网络问题未能及时发送ack，RabbitMQ可能会重新投递消息。消费者再次接收时，可能已经处理过该消息，导致重复消费。
- 常见的业务问题包括：
- 库存扣减：执行两次扣减操作，导致库存数量错误。
- 余额扣款：重复扣款可能导致余额错误。
如何确保业务幂等性
为了避免消息重复消费带来的问题，确保业务幂等性是关键。具体做法有：
- 业务本身具备幂等性：如查询操作或删除操作，这些操作天生幂等。查询一次与查询多次结果一致，删除操作只要删除成功，后续重复删除无影响。
- 通过业务判断实现幂等性：对于非幂等业务，如扣款或库存扣减，我们需要加入额外的业务逻辑来避免重复处理。
解决方案
方案一：唯一消息ID
通过为每条消息生成唯一的ID，可以确保消费者识别消息是否被重复消费。具体步骤如下：
- 发送方为每条消息生成唯一ID，并将其放入消息头部。
- 消费者在处理消息时，将消息ID与数据库中的记录比对，如果ID已存在，则认为该消息是重复消费，跳过处理。
实现步骤：
- 在Spring AMQP中，配置
MessageConverter来自动为每条消息生成唯一ID。 - 消费者收到消息时，首先检查消息ID是否已存在于数据库中，如果存在则跳过处理。
@Bean
public MessageConverter messageConverter() {
// 1.定义消息转换器
Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jjmc.setCreatteMessageIds(true);
return jjmc;
}
方案二：业务逻辑判断
基于业务的特点，判断当前操作是否可以执行，从而避免重复执行。例如，在支付业务中，我们可以通过判断订单的状态来决定是否需要执行支付操作。
示例：在处理订单支付时，订单的状态应为“未支付”才能进行支付操作。如果订单状态已经是“已支付”或“已退款”，则认为该消息是重复的，直接跳过处理。
具体步骤：
- 查询订单状态，判断是否为“未支付”。
- 如果是“未支付”，执行支付操作并更新订单状态；否则，跳过此操作。
异常处理与重试机制
在实际生产环境中，网络问题可能导致消息未确认或处理失败，因此需要设计有效的重试机制。Spring AMQP支持消费者确认机制和失败重试机制，常见的策略有：
- 发送者确认机制：确保消息成功投递到RabbitMQ。
- 消费者确认机制：确保消费者成功处理消息。
- 失败重试策略：如果消息处理失败，可以配置重试次数与延迟，以确保消息最终被成功处理。
消息的持久化与可靠性
为了保证消息不丢失，RabbitMQ支持消息持久化。消费者确认机制（ack）和消息持久化可以保证在系统故障或重启后，消息能够正确传递给消费者。
面试问题回顾
在实际的面试中，可能会问到如何保证微服务间状态一致性（如支付服务与交易服务之间的状态一致性）。回答时可以提到：
- 使用消息队列（如RabbitMQ）来实现异步通知，避免同步调用的性能瓶颈和耦合。
- 为了确保消息投递的可靠性，采用生产者确认机制、消费者确认机制以及消息持久化策略。
- 对于重复消费问题，采用业务逻辑判断或唯一消息ID来保证幂等性。
总结
解决RabbitMQ中的消息重复消费问题，关键是保证业务逻辑的幂等性。通过唯一ID或基于业务的判断逻辑，可以确保消息的重复消费不会导致不一致或错误的业务状态。同时，借助消息确认机制和重试机制，可以确保消息投递和处理的可靠性。
代码示例
配置消息ID生成
@Bean
public MessageConverter messageConverter() {
Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
// 自动生成唯一消息ID
converter.setCreateMessageId(true);
return converter;
}
消费者接收消息时获取消息ID
@RabbitListener(queues = "simpleQueue")
public void receiveMessage(Message message) {
String messageId = message.getMessageProperties().getMessageId();
// 判断消息是否重复消费
if (messageAlreadyProcessed(messageId)) {
return; // 跳过重复消息
}
// 处理消息
processMessage(message);
}
订单支付状态判断
public void processPayment(Order order) {
// 查询订单状态
Order currentOrder = orderService.getOrderById(order.getId());
// 判断订单状态是否为"未支付"
if (currentOrder.getStatus() != OrderStatus.PENDING) {
return; // 订单已支付或已退款，不需要重复支付
}
// 执行支付操作
orderService.processPayment(order);
}