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

RabbitMQ确保消息可靠性

article 2025/1/16 6:33:56

消息丢失的可能性

在这里插入图片描述
支付服务先扣减余额和更新支付状态（这俩是同步调用），然后通过RabbitMq异步调用支付服务更新订单状态。但是有些情况下，可能订单已经支付 ，但是更新订单状态却失败了，这就出现了消息丢失。

  1. 发送者在发送的过程中出现了网络故障
  2. RabbitMQ在发送消息的过程中出现了问题
  3. 消费者在更新订单状态的时候出现了问题

发送者的可靠性

发送者确认机制需要与MQ进行通信和确认，会影响消息发送的效率且一般出现的概率极低，所以一般不用这个。

方法1. 发送者重连

确保发送者与MQ之间连接的可靠性。有的时候由于网络波动，可能出现发送者连接MQ失败的情况，这个配置是关闭的，可以开启连接失败后的重连机制：

spring:
  rabbitmq:
    connection-timeout: 1s # 设置MQ的连接超时时间
    template:
      retry:
        enabled: true # 开启超时重试机制(默认是false)
        initial-interval: 1000ms # 失败后的初始等待时间
        multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
        max-attempts: 3 # 最大重试次数

【注】：当网络不稳定时，利用重试机制可以提高消息发送的成功率，但是SpringAMQP提供的重试机制是阻塞式的重试，如果需要多次重试等待，当前线程被阻塞，会影响性能。
如果对业务性能有要求，建议禁用重试机制，如果一定要使用，要合理的配置等待时常和重试次数，或使用异步线程来执行发送消息的代码。

方法2. 发送者确认

确保消息发送的可靠性。SpringAMQP提供了Publisher Confirm和Publisher Return两种机制，开启确认机制后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者，返回的结果有以下几种情况：

  • 消息投递到MQ，但是路由失败，此时通过PublisherReturn返回路由异常信息，然后返回ACK，告知投递成功。例如：
    • 消息发送给图中的exchange1，但是RoutingKey写错了，没有匹配到正确的队列，也会导致路由失败。
    • 消息发送给图中的exchange2，但是它底下没有绑定新的队列，就会导致路由失败。
  • 临时消息【不需要往磁盘做持久化的消息】投递到MQ，并入队成功，返回ACK，告知投递成功。
  • 持久消息投递到MQ，并入队完成持久化，返回ACK，告知投递成功。
  • 其他情况都会返回NACK，告知投递失败。
    在这里插入图片描述
    步骤：
  1. 在发送方publisher所在的微服务的application.yml中配置：
spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
    publisher-returns: true # 开启publisher return机制

publisher-confirm-type有三种模式：

  • none：关闭confirm机制
  • simple：同步阻塞等待MQ回执消息
  • correlated：MQ异步回调方式返回回执消息（常用）
  1. 开启回调机制：每个RabbitTemplate只能配置一个ReturnCallback，在发送者publisher所在的项目启动时配置即可。
@Configuration
@RequiredArgsConstructor
@Slf4j
public class MqConfig {
    private final RabbitTemplate rabbitTemplate;
    @PostConstruct // 在Bean初始化完成后调用这个方法（只会调用一次）
    public void init() {
    	// 返回ACK，但是此时路由失败，就会走这个方法
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            log.error("监听到了消息return callback");
            log.debug("exchange: {}", returnedMessage.getExchange());
            log.debug("routingKey: {}", returnedMessage.getRoutingKey());
            log.debug("message: {}", returnedMessage.getMessage());
            log.debug("replyCode: {}", returnedMessage.getReplyCode());
            log.debug("replyText: {}", returnedMessage.getReplyText());
        });
    }
}
  1. 开启消息确认机制：发送消息、指定消息ID、每次发送消息都需要配置一个ConfirmCallback
public void testConfirmCallback() {
  CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
  cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
      // 【几乎不可能发生】Future发生异常时的处理逻辑
      @Override
      public void onFailure(Throwable ex) {
          log.error("spring ampq处理确认结果异常", ex);
      }

      // 成功，拿到MQ结果，判断是ACK还是NACK
      @Override
      public void onSuccess(CorrelationData.Confirm result) {
          if(result.isAck()) {
              // ACK
              log.debug("收到ACK，消息发送成功");
          }else {
              // NACK
              log.debug("收到NACK，消息发送失败，失败原因：{}",result.getReason());
          }
      }
  });
  rabbitTemplate.convertAndSend("hmall.direct", "blue", "hello world", cd); // 发送消息
}

MQ的可靠性

RabbitMQ一般会将收到的信息保存到内存（速度快）中，降低消息收发的延迟，这样会导致：

  1. MQ宕机，内存中的消息会丢失。
  2. 内存空间有限，消费者故障或处理过慢，会导致消息积压，引发MQ阻塞。
    在这里插入图片描述

【案例】发送者往MQ发消息，MQ会把数据保存到内存中，如果内存满了，MQ就会把一部分数据迁移到磁盘中暂时进行持久化存储，移动到磁盘的这段时间发送者发送的消息就会产生丢失。

方法1. 数据持久化

数据持久化就是把数据持久化到磁盘，但是不是向上边那个案例，等满了再去持久化（被动），而是提前进行持久化。

  1. 交换机的持久化（默认开启的）
    在这里插入图片描述
  2. 队列的持久化（默认开启的）
    在这里插入图片描述
  3. 消息持久化（默认是非持久的）
    在发送消息的时候设定的
    在这里插入图片描述
    【案例】：比较一下持久化和非持久化的性能。
    发100w条消息给MQ：
    这是非持久化的方式：使用纯内存的方式存储，每次内存满之后，MQ就会把消息写到磁盘中，此时就会出现阻塞状态，处理速度降低到0
    在这里插入图片描述
    【问题】可能出现消息丢失和MQ阻塞
    【解决办法】使用持久化的方式：
public void testSendPersistentMsg() {
    // 自定义构建消息
    Message msg = MessageBuilder.withBody("hello world".getBytes(StandardCharsets.UTF_8)) // 消息体
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 投递模式（持久化）
            .build();
    for (int i = 0; i < 100000; ++i) {
        rabbitTemplate.convertAndSend("simple.queue", msg);
    }
}

Mq并没有阻塞，每发一条消息就赶紧把它存到磁盘中，和纯内存方式相比，不会有个中断的过程。
在这里插入图片描述

方法2. Lazy Queue（推荐）

【问题】：由于使用了消息持久化的方式，发到MQ的消息不仅要到内存，还要在磁盘中写一份，这会导致整体的并发能力下降
【特征】：

  • 接收到消息后直接入磁盘，不再存储到内存
  • 在写磁盘的时候也对写入磁盘的操作进行一些优化，比传统的写操作高很多
  • 消费者要消费消息时，才会从磁盘中读取并加载到内存
    • 【问题】：可能会影响消费者处理消息的速度
    • 【解决】：可以提前缓存部分消息到内存，最多2048条

控制台声明Lazy Queue队列

在这里插入图片描述

Java代码添加

声明Bean
@Bean
public Queue lazyQueue(){
    return QueueBuilder
            .durable("lazy.queue")
            .lazy() // 开启Lazy模式
            .build();
}
@RabbitListener注解
@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy") // 开启Lazy模式
))
public void listenLazyQueue(String msg){
    log.info("接收到 lazy.queue的消息：{}", msg);
}

消费者的可靠性

消费者确认机制

为了确认消费者是否成功处理消息，当消费者处理消息结束后，应该向MQ发送一个回执，告知MQ自己的消息处理状态。有如下几种消息处理状态：

  • ack：处理消息成功，RabbitMQ从队列中删除该消息
  • nack：消息处理失败，RabbitMQ需要再次投递消息
  • reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息【在处理的过程中，发现消息的内容有问题，没有重试的必要，直接拒绝就行】

【注意】：不管是哪种情况，都应该等消息处理完后得到结果再返回，不要一拿到消息就返回
返回消息处理状态的过程，类似于处理事务，事务处理成功，返回ACK；处理失败，返回NACK

SpringAMQP允许通过在消费者的配置文件选择ACK的处理方式，有三种：

  • none：不处理，消息投递给消费者后立刻ack，消息会立刻从MQ中删除，别用
  • manual：手动模式，需要在业务代码中调用api，发送ack或reject，存在业务入侵，但是更灵活。
  • auto：自动模式，利用AOP对消息处理逻辑进行了环绕增强
    • 业务处理正常：自动返回ack
    • 业务处理异常：自动返回nack
    • 消息处理或校验异常【MessageConversionException】：自动返回reject
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto# 不做处理

失败重试策略

在消费者出现异常时，利用本地重试，而不是无限的重新入队到mq，可以在消费者的yaml文件中添加配置来开启重试机制。

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初识的失败等待时长为1秒
          multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数（达到最大重试次数后，MQ会把消息丢弃）
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

【问题】：在开启重试模式后，重试次数耗尽，如果消息仍然失败，默认会把消息进行丢弃。
【解决】：因此需要有MessageRecoverer接口来处理，包含三种不同的实现：

  • RejectAndDontRequeueRecoverer（默认）：重试耗尽后，直接reject，丢弃消息。
  • ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队。
  • RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机 。
    在这里插入图片描述

修改失败重试策略为RepublishMessageRecoverer

  1. 定义接收失败的交换机、队列、
  2. 定义RepublishMessageRecoverer
@Configuration
public class ErrorMessageConfiguration {
    // 定义接收失败的交换机
    @Bean
    public DirectExchange errorExchange() {
        return new DirectExchange("error.direct");
    }

    // 定义接收失败的队列
    @Bean
    public Queue errorQueue() {
        return new Queue("error.queue");
    }

    // 定义绑定关系
    @Bean
    public Binding errorQueueBinding() {
        return BindingBuilder.bind(errorQueue())
                .to(errorExchange())
                .with("error");
    }
    
    // 定义失败处理策略
    @Bean
    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

业务幂等性

f(x) = f(f(x))，指同一个业务，执行一次或多次对业务状态的影响是一致的。

  • 幂等业务：查询业务、删除业务
  • 非幂等业务：用户下单需要扣减库存、用户退款业务需要恢复余额

方案1. 唯一消息id

给每个消息设置一个唯一id，利用id区分是否是重复消息：

  • 每条消息都生成一个唯一id，与消息一起投递给消费者
  • 消费者接收到消息后处理自己的业务，业务处理成功后将消息id保存到数据库中
  • 如果下次又收到相同消息，去数据库查询判断是否存在，存在则视为重复消息放弃处理
  1. 在发送方配置Bean用来自动创建消息id
@Configuration
public class MqConfig {
    @Bean
    public MessageConverter messageConverter() {
        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
        converter.setCreateMessageIds(true); // 配置自动创建消息id
        return converter;
    }
}
  1. 在接收方接收消息id
@RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(Message msg) { // 使用字符串发送，就用字符串接收
        log.info("监听到simple.queue的消息：{}", msg);
        log.info("消息id：{}", msg.getMessageProperties().getMessageId());
        // throw new RuntimeException("故意的");
    }

在这里插入图片描述

方案2. 业务判断（常用）

结合业务逻辑，基于业务本身做判断。
【案例】：当用户下单成功后，通过MQ通知交易服务来修改订单状态为已支付（这里记作消息1），修改成功后交易服务返回ACK给MQ，此时出现了网络的故障，MQ没有收到交易服务发送的ACK，MQ认为交易服务宕机，消息又重新入队。
就在此刻，用户点击了申请退款，直接向交易服务修改订单状态为退款中（这个操作没有走MQ，此时订单状态是退款中，但是消息1还在消息队列中）。
此时网络恢复了，MQ又将消息1发送给交易服务，此时交易服务又把订单状态标记为已支付（订单申请退款中的状态又被覆盖了）。
在这里插入图片描述
【解决】：通知来的时候，先判断订单的状态，再进行操作。

@Component
@RequiredArgsConstructor
public class PayStatusListener {
    private final IOrderService orderService;
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "trade.pay.success.queue", durable = "true"),
            exchange = @Exchange(name = "pay.direct"),
            key = "pay.success"
    ))
    public void listenPaySuccess(Long orderId) {
        // 1.查询订单
        Order order = orderService.getById(orderId);
        // 2.判断订单状态是否为未支付
        if(order == null || order.getStatus() != 1) {
            // 不做处理
            return;
        }
        // 3.标记订单状态为已支付
        orderService.markOrderPaySuccess(orderId);
    }
}

在这里插入图片描述

延迟消息

延迟消息：发送者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
延迟任务：设置在一定时间后才执行的任务。
在这里插入图片描述

方案1. 死信交换机

当一个队列中的消息满足下列情况之一的，就会成为死信：

  • 消费者使用basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false。
  • 消息是一个过期消息（达到队列设置的过期时间 或 消息本身设置的过期时间），超时无人消费。
  • 要投递的队列消息堆积满了，最早的消息可能成为死信。

队列通过dead-letter-exchange属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中，这个交换机就叫做死信交换机（DLX）。
在这里插入图片描述

  1. 声明死信队列、死信交换机、它们之间的绑定关系：
@RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "dlx.queue", durable = "true"), // 死信队列
            exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT), // 死信交换机
            key = {"hi"} 
    ))
    public void listenDlxQueue(String msg) {
        log.info("消费者监听到dlx.queue的消息: " + msg);
    }
  1. 声明普通队列、普通交换机、它们之间的绑定关系，并把队列绑定到死信交换机上（此时就不需要把它绑定消费者了）：
@Configuration
public class NormalConfiguration {
    @Bean
    public DirectExchange normalExchange() { // 普通交换机
        return ExchangeBuilder.directExchange("normal.direct").build();
    }
    @Bean
    public Queue normalQueue() { // 普通队列
        return QueueBuilder
                .durable("normal.direct") // 队列名字
                .deadLetterExchange("dlx.direct") // 死信交换机名字
                .build();
    }
    @Bean
    public Binding normalQueueBinding(Queue normalQueue, DirectExchange normalExchange) { // 绑定关系
        // 把队列绑定到交换机
        return BindingBuilder
                .bind(normalQueue) // 队列
                .to(normalExchange) // 交换机
                .with("hi");// 这里绑定关系要和普通队列的绑定关系保持一致
    }
}
  1. 发送延迟消息：
@Test
public void testSendDelayMsg() {
    rabbitTemplate.convertAndSend("normal.direct", "hi", "hello world", message -> {
        // 当消息被转成Message对象后，还可以进一步做加工
        message.getMessageProperties().setExpiration("10000"); // 设置消息过期时间(10s)
        return message;
    });
}

【注】：normal.direct和normal.queue之间绑定的BindingKey 与 dlx.direct和dlx.queue之间绑定的BindingKey要一致

方案2. 延迟消息插件DelayExchange（推荐）

这个插件可以将普通交换机改造为支持延迟消息功能的交换机，当消息投递到交换机后，可以暂存一段时间，到后期再投递到队列。

一、安装插件

  1. 插件下载地址：DelayExchange
  2. 需要把插件放在RabbitMQ插件目录对应的数据卷下
docker volume inspect mq-plugins

在这里插入图片描述
在这里插入图片描述

  1. 执行命令，安装插件
docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

二、使用插件

  1. 声明延迟交换机：只要设置delay的属性为true即可
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue", durable = "true"),
        exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT), // 只要设置一个delayed属性为true即可
        key = {"hi"}
))
public void listenDelayQueue(String msg) {
    log.info("消费者监听到delay.queue的消息: " + msg);
}
  1. 发送延迟消息：通过消息头x-delay来设置过期时间
@Test
public void testSendDelayMsgByPlugin() {
    rabbitTemplate.convertAndSend("delay.direct", "delay", "hello world", message -> {
        message.getMessageProperties().setDelay(10000);// 添加延迟消息属性
        return message;
    });
}

延迟消息的实现需要记录消息的过期时间，计时的时钟需要依赖cpu，是个cpu密集型任务。因此使用延迟消息时，需要避免同一时刻在mq里存在大量的延迟消息（尽可能地让延迟消息的延迟时间不要太长）。


http://www.kler.cn/a/504644.html

相关文章：

  • 微信小程序集成Vant Weapp移动端开发的框架
  • 解决 VSCode 调试时 Python 文件出现相对路径报错问题‘FileNotFoundError’
  • 个人主页搭建全流程(Nginx部署+SSL配置+DCDN加速)
  • CryptoMamba：利用状态空间模型实现精确的比特币价格预测
  • [微服务]redis数据结构
  • 行业案例：高德服务单元化方案和架构实践
  • LSTM火灾温度预测（Pytorch版本）
  • 【JavaScript】基础内容，HTML如何引用JavaScript, JS 常用的数据类型
  • Ubuntu创建AndroidStudio快捷启动图标
  • HTML拖拽功能（纯html5+JS实现）
  • 极限竞速 地平线5“d3dx12_43.dll”文件丢失或错误导致游戏运行异常如何解决？windows系统DLL文件修复方法
  • 智汇云舟参编《城市轨道交通安全防范系统技术要求》国标正式发布
  • 【原创】大数据治理入门（2）《提升数据质量：质量评估与改进策略》入门必看 高赞实用
  • 面试之《new关键字》
  • python创建pdf水印，希望根据文本长度调整水印字体大小，避免超出页面
  • [Qt]常用控件介绍-输入类控件-QSpinBox、DateTimeEdit、Dial、Silder控件
  • C++笔记：打包独立运行的exe（在静态库中使用MFC）
  • 使用yarn命令创建Vue3项目
  • Microsoft Sql Server 2019 触发器
  • 利用 rclone 挂载华为云 OBS 到本地
  • 图-岛屿-dfs
  • 什么是docker？关于docker容器的全面详细介绍
  • Spring MVC流程一张图理解
  • 获取文章列表功能
  • LeetCode热题100-有效的括号【JavaScript讲解】
  • 常见好用的PHP CMS开源系统有哪些？