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);