欢迎光临
RabbitMQ——消息应答和持久化
   

RabbitMQ——消息应答和持久化

文章目录

  • RabbitMQ——消息应答和持久化
    • 1、消息应答
      • 1.1、概念
      • 1.2、手动应答示例
      • 2、持久化

        RabbitMQ——消息应答和持久化

        1、消息应答

        1.1、概念

        概念

        消息应答机制是指消费者在消费消息后向 RabbitMQ 确认(acknowledge)已经成功处理了消息。

        这个机制有助于确保消息在被消费者处理后被正确地从队列中移除,从而防止消息的丢失。

        两种消息应答机制

        1、自动应答(Auto Acknowledgment): 在自动应答模式下,一旦消息被消费者接收,RabbitMQ 会立即将消息标记为已被消费,而不需要消费者明确地向 RabbitMQ 发送确认。

        这种模式下,消息被认为已经成功处理,即使消费者在处理消息的过程中发生错误,消息也会从队列中删除。

        开启自动应答: 在消费者订阅队列时,设置 autoAck 参数为 true。

        channel.basicConsume(queueName, true, consumer);
        

        优点:简单,不需要手动确认。

        适用场景:对消息的处理时机和可靠性要求不高,可以容忍一定程度的消息丢失。

        2、手动应答(Manual Acknowledgment): 在手动应答模式下,消费者在处理完消息之后,需要向 RabbitMQ 发送明确的确认信号,告诉 RabbitMQ 可以安全地删除这条消息了。

        这种模式下,消费者需要手动调用确认方法。

        • 关闭自动应答: 在消费者订阅队列时,设置 autoAck 参数为 false。

          channel.basicConsume(queueName, false, consumer);
          
        • 手动确认消息: 在消费者处理完消息后,手动向 RabbitMQ 发送确认信号。

          channel.basicAck(deliveryTag, false);
          

          优点:

          1、更精细的控制消息的处理时机和可靠性,确保消息在成功处理后才被确认。

          2、可以批量应答,并且减少网络拥堵

          适用场景:对消息的可靠性传递有较高要求,需要在消息处理成功后才确认消息,以避免消息丢失。

          1.2、手动应答示例

          生产者:

          package com.weipch.rabbitmq.three;
          import com.rabbitmq.client.Channel;
          import com.rabbitmq.client.MessageProperties;
          import com.weipch.rabbitmq.utils.RabbitMqUtils;
          import java.nio.charset.StandardCharsets;
          import java.util.Scanner;
          public class Task02 {
              public static final String TASK_QUEUE_NAME = "ack_queue";
              public static void main(String[] args) throws Exception {
                  Channel channel = RabbitMqUtils.getChannel();
                  boolean durable = true;
                  channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null);
                  Scanner scanner = new Scanner(System.in);
                  while (scanner.hasNext()) {
                      String message = scanner.next();
          //            要求生产者发送消息为持久化消息(要求保存到磁盘中)
                      channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
                      System.out.println("生产者发出消息:" + message);
                  }
              }
          }
          

          消费者:

          package com.weipch.rabbitmq.three;
          import com.rabbitmq.client.Channel;
          import com.rabbitmq.client.DeliverCallback;
          import com.weipch.rabbitmq.utils.RabbitMqUtils;
          import com.weipch.rabbitmq.utils.SleepUtils;
          public class Worker03 {
              public static final String TASK_QUEUE_NAME = "ack_queue";
              public static void main(String[] args) throws Exception {
                  Channel channel = RabbitMqUtils.getChannel();
                  System.out.println("C1等待接收消息处理时间较短");
                  DeliverCallback deliverCallback=(consumerTag, message)->{
                      SleepUtils.sleep(1);
                      System.out.println("接收到的消息:"+new String(message.getBody()));
                      /*手动应答
                      1.deliveryTag: 消息的唯一标识符,用于确认具体的哪条消息已经被处理
                      2.是否批量应答 false:不批量应答信道中的消息 true:批量
                      * */
                      channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                  };
                  boolean autoAck = false;
                  channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, (consumerTag -> System.out.println(consumerTag+"消息被取消消费接口回调逻辑")));
              }
          }
          

          2、持久化

          队列持久化

          // 声明持久化的队列
          boolean durable = true;
          channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null);
          

          消息持久化

          //要求生产者发送消息为持久化消息(要求保存到磁盘中)
          //将消息的属性设置为 MessageProperties.PERSISTENT_TEXT_PLAIN,
          channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
          

          即使消息被标记为持久化,也不能绝对保证消息在每个可能的故障场景下都不会丢失。例如:操作系统在写入磁盘时可能会使用缓存,而不是立即将数据写入磁盘。在极端情况下,如果发生硬件故障或操作系统崩溃,缓存中的数据可能还未来得及写入磁盘,导致数据丢失。

          RabbitMQ的分发机制:

          • 轮询分发(不公平分发):这是RabbitMQ的默认分发策略。在这种模式下,消息会按顺序逐个发送给消费者。每条消息在被确认(acknowledged)后,即被从队列中移除,然后下一条消息会被发送给下一个消费者。这种机制确保了消息的公平分配,而不考虑消费者的处理能力或者负载情况。这可能导致某些消费者处理较快,而其他消费者处理较慢,从而影响整体的消息处理效率。

          • 公平分发(能者多劳):这种模式下,消息的分发会根据消费者的处理能力来决定。这意味着处理速度快的消费者会接收到更多的消息,而处理速度慢的消费者则会接收到较少的消息。这种方式旨在提高整体的处理效率,确保每个消费者都能根据自身的处理能力得到合理数量的消息。

            // basicQos设置为1是公平分发
            int prefetchCount = 1;
            channel.basicQos(prefetchCount);
            

            预取值

            RabbitMQ中的预取(prefetch)是指在消费者从队列获取消息之前,队列可以推送多少条消息给消费者。设置预取值的主要目的是控制消费者获取消息的速率,防止某个消费者过快地消费消息,导致其他消费者无法及时处理消息。

            在RabbitMQ中,可以通过basicQos方法来设置预取值。

            // 设置预取值,这里设置为一次只推送5条消息给消费者
            int prefetchCount = 5;
            channel.basicQos(prefetchCount);
            
             
打赏
版权声明:本文采用知识共享 署名4.0国际许可协议 [BY-NC-SA] 进行授权
文章名称:《RabbitMQ——消息应答和持久化》
文章链接:https://goodmancom.com/wl/175661.html