Spring AMQP-保证消费者消息的可靠性

news2025/1/13 4:52:15

为什么要保证消息的可靠性？

当MQ向消费者发送一个消息之后需要得到消费者的状态，因为消息并不一定就真的被消费者给消费了，可能在消费的过程中出现了一些意外，比如

1. 网络问题

2. 消息转换有问题

3. 消费者本身的业务处理有问题

消费者确认机制

消费者消息处理状态：

ack:消息成功接收，并且成功被处理，MQ将此消息删除
nack：消息处理失败，需要MQ重新发送消息
reject：消息处理失败并且拒绝该消息，MQ将此消息删除

由于消息回执的处理代码比较统一，因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式，有三种模式：

none：不处理。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用

manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活

auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack. 当业务出现异常时，根据异常判断返回不同结果：

如果是业务异常，会自动返回nack；

如果是消息处理或校验异常，自动返回reject;

在配置文件中通过下面的配置即可设置ACK的处理方式
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: c # none 默认 auto 自动确认 manual 手动确认 

消费者重试机制

消费者接收了一个消息，但是在处理的过程中出现异常了，那么AMQP会不断的重试，直到把资源占完然后崩掉，这个时候就必须要设置重试机制，限制重试的次数，避免无限制重试。
spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初识的失败等待时长为1秒
          multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false
测试

先别配置重试机制，然后在需要在接收消息的地方手动抛出一个异常，查看控制台就会看见消费者在尝试不断的获取消息，但是一直获取不到无限制的重试
  @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "fanout.hamll.query1", // 队列名称
                    durable = "true"), // 是否持久
            exchange = @Exchange(name = "fanout.hamll", type = ExchangeTypes.FANOUT) // 交换机名称
    ))
    public void query2(String message) {
        System.err.println("fanout.hamll.query1 消息内容：" + message); 
        throw  new RuntimeException("故意的错误"); // 抛出异常
    }
配置好重试之后到了三次就会直接停止，这样子就很好的减少了系统资源的消耗

业务的幂等性判断

什么是幂等性？

在Java领域，幂等性是指同一个请求，不管发送多少次执行的结构都是一样的。

比如支付和交易，支付成功之后通知交易服务修改状态。在交易服务需要查询订单并判断订单的状态，这样子不管同一个订单重复发起多少次请求，都不会对业务的结果造成影响。

MQ保证消息的幂等性

MQ中的幂等是说，不管消息是否被重复消费，都不会对业务造成影响、处理的结果都是一致的。

MQ实现业务幂等性

为每个消息都创建一个唯一的MessageId在操作的时候将其存入数据库，然后在进行判断消息是否存在，存在就直接跳过业务的处理，不存在就继续操作。
    @Bean
    public MessageConverter messageConverter(){
        // 1.定义消息转换器
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
        jackson2JsonMessageConverter.setCreateMessageIds(true);
        return jackson2JsonMessageConverter;
    }
业务实现幂等性

在业务的操作中，比如支付和交易服务，支付成功之后会通知交易服务修改订单的状态，而在交易服务应该做判断，判断该订单的状态是否未未支付。如果是未支付就继续处理接下来的业务，否则就直接结束。
package com.hmall.trade.listener;

import com.hmall.trade.domain.po.Order;
import com.hmall.trade.service.IOrderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class PayStatusListener {
    @Autowired
    IOrderService orderService;
    @RabbitListener(bindings = @QueueBinding(
                value = @Queue(value = "trade.pay.success.queue", durable = "true"),
            exchange = @Exchange(value = "pay.direct", type = ExchangeTypes.DIRECT),
            key = "pay.success"
    ))
    public void paySuccess(Long orderId) {
        log.info("支付成功，订单号：{}", orderId);
        //查询当前订单 判断幂等性
        Order order = orderService.getById(orderId);
        //判断状态以及对象是否存在
        if (order == null || order.getStatus() != 1) {
            return;
        }
        orderService.markOrderPaySuccess(orderId);
    }
}