RabbitMQ——消息应答和持久化

文章目录

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、手动应答示例

生产者：

packagecom.weipch.rabbitmq.three;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.MessageProperties;importcom.weipch.rabbitmq.utils.RabbitMqUtils;importjava.nio.charset.StandardCharsets;importjava.util.Scanner;publicclassTask02{publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";publicstaticvoidmain(String[] args)throwsException{Channel channel =RabbitMqUtils.getChannel();boolean durable =true;
        channel.queueDeclare(TASK_QUEUE_NAME, durable,false,false,null);Scanner scanner =newScanner(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);}}}

消费者：

packagecom.weipch.rabbitmq.three;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importcom.weipch.rabbitmq.utils.RabbitMqUtils;importcom.weipch.rabbitmq.utils.SleepUtils;publicclassWorker03{publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";publicstaticvoidmain(String[] args)throwsException{Channel channel =RabbitMqUtils.getChannel();System.out.println("C1等待接收消息处理时间较短");DeliverCallback deliverCallback=(consumerTag, message)->{SleepUtils.sleep(1);System.out.println("接收到的消息："+newString(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);