【RabbitMQ笔记07】消息队列RabbitMQ七种模式之Publisher Confirms发布确认模式

这篇文章，主要接收消息队列RabbitMQ七种模式之Publisher Confirms发布确认模式。

目录

一、消息队列

1.1、发布确认模式

1.2、案例代码

（1）引入依赖

（2）编写生产者【消息确认--单条确认】

（3）编写生产者【消息确认--批量确认】

（4）编写生产者【消息确认--异步确认】

---

一、消息队列

1.1、发布确认模式

RabbitMQ消息队列中，生产者发送消息给RabbitMQ的时候，可能会出现发送失败的情况，如果不进行处理，此时这一条消息就将丢失。如何确保生产者一定能够将消息发送到RabbitMQ里面呢？？？

RabbitMQ提出了一种发布确认模式，这种模式大致思想是：生产者发送消息给RabbitMQ时候，如果RabbitMQ正确接收到消息后，需要发给一个ACK标识给生产者，生产者接收到ACK标记后，就可以确认这一条消息发送成功啦。如果生产者没有接收到ACK标识，则可以重复发送这一条消息给RabbitMQ，这就可以确保消息不丢失。

发布确认模式有三种实现，分别是：逐条确认机制、批量确认机制、异步确认机制。

1.2、案例代码

（1）引入依赖

<!-- 引入 RabbitMQ 依赖 -->
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.16.0</version>
</dependency>

（2）编写生产者【消息确认--单条确认】

• 生产者发送消息的时候，需要调用【confirmSelect()】方法开启消息确认机制。
• 生产者将消息发送完成之后，需要调用【waitForConfirms()】方法，阻塞等待RabbitMQ消息队列返回ACK标识。这个方法返回一个boolean类型，true表示RabbitMQ接收消息成功，false表示接收失败。
• 【waitForConfirms()】方法还可以指定一个超时时间，如果在这个超时时间里面RabbitMQ还没有返回ACK标识，那么该方法将抛出一个InterruptedException中断异常。

package com.rabbitmq.demo.confirm;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** @version 1.0.0* @Date: 2023/2/25 16:23* @Copyright (C) ZhuYouBin* @Description: 消息生产者*/
public class Producer {public static void main(String[] args) {// 1、创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2、设置连接的 RabbitMQ 服务地址factory.setHost("127.0.0.1"); // 默认就是本机factory.setPort(5672); // 默认就是 5672 端口// 3、获取连接Connection connection = null; // 连接Channel channel = null; // 通道try {connection = factory.newConnection();// 4、获取通道channel = connection.createChannel();// TODO 开启消息确认机制channel.confirmSelect();// 5、声明 Exchange，如果不存在，则会创建String exchangeName = "exchange_direct_2023";channel.exchangeDeclare(exchangeName, "direct");// 6、发送消息for (int i = 0; i < 10; i++) {// 路由键唯一标识String routingKey = "error";if (i % 3 == 0) {routingKey = "info";} else if (i % 3 == 1) {routingKey = "warn";}String message = "这是发布确认模式，发送的第【" + (i+1) + "】条【" + routingKey + "】消息数据";channel.basicPublish(exchangeName, routingKey, null, message.getBytes());// 等待RabbitMQ返回ACK标识boolean wait = channel.waitForConfirms();System.out.println("RabbitMQ是否接收成功: " + wait);if (!wait) {// 消息发送失败，则可以重新发送channel.basicPublish(exchangeName, routingKey, null, message.getBytes());}}} catch (Exception e) {e.printStackTrace();} finally {if (null != channel) {try {channel.close();} catch (Exception e) {}}if (null != connection) {try {connection.close();} catch (Exception e) {}}}}
}

（3）编写生产者【消息确认--批量确认】

• 前一种方式，是一条消息就调用一次【waitForConfirms()】方法，阻塞等待RabbitMQ的ACK确认标识。
• 但是这种方式是非常耗时的，当需要发送的消息非常多的时候，会严重影响系统性能，所以为了解决这个问题，提出了批量确认的方法。
• 批量确认调用【waitForConfirmsOrDie()】方法，此时会等待一批消息的ACK确认标识，如果这一批消息中存在一个消息没有被RabbitMQ成功接收，此时该方法将抛出一个【IOException】异常。
• 所以，可以通过捕获IOException异常来判断消息是否发送成功。
• 这种方式的缺点：当一批消息出现失败的情况时候，我们没办法知道是哪一条消息失败了，只能够重新将这一批消息重新发送。

package com.rabbitmq.demo.confirm;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;/*** @version 1.0.0* @Date: 2023/2/25 16:23* @Copyright (C) ZhuYouBin* @Description: 消息生产者*/
public class ProducerBatch {public static void main(String[] args) {// 1、创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2、设置连接的 RabbitMQ 服务地址factory.setHost("127.0.0.1"); // 默认就是本机factory.setPort(5672); // 默认就是 5672 端口// 3、获取连接Connection connection = null; // 连接Channel channel = null; // 通道try {connection = factory.newConnection();// 4、获取通道channel = connection.createChannel();// TODO 开启消息确认机制channel.confirmSelect();// 5、声明 Exchange，如果不存在，则会创建String exchangeName = "exchange_direct_2023";channel.exchangeDeclare(exchangeName, "direct");// 6、发送消息int batchSize = 3;int count = 0;for (int i = 0; i < 10; i++) {// 路由键唯一标识String routingKey = "error";if (i % 3 == 0) {routingKey = "info";} else if (i % 3 == 1) {routingKey = "warn";}String message = "这是发布确认模式，发送的第【" + (i+1) + "】条【" + routingKey + "】消息数据";channel.basicPublish(exchangeName, routingKey, null, message.getBytes());// 批量确认if (count == batchSize) {// 等待RabbitMQ返回ACK标识channel.waitForConfirmsOrDie();count = 0;}count++;}} catch (IOException e) {System.out.println("消息发送失败啦");} catch (Exception e) {e.printStackTrace();} finally {if (null != channel) {try {channel.close();} catch (Exception e) {}}if (null != connection) {try {connection.close();} catch (Exception e) {}}}}
}

（4）编写生产者【消息确认--异步确认】

• 异步确认在消息发送之后，调用【addConfirmListener()】方法，该方法介绍两个参数，第一个参数是成功接收到ACK标识的回调方法，第二个参数是失败接收到NACK标识的回调方法。
• 注意：一定要先调用【addConfirmListener()】监听方法，然后再发送消息，如果两者顺序反了，则监听方法不生效。

package com.rabbitmq.demo.confirm;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;/*** @version 1.0.0* @Date: 2023/2/25 16:23* @Copyright (C) ZhuYouBin* @Description: 消息生产者*/
public class ProducerAsync {public static void main(String[] args) {// 1、创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2、设置连接的 RabbitMQ 服务地址factory.setHost("127.0.0.1"); // 默认就是本机factory.setPort(5672); // 默认就是 5672 端口// 3、获取连接Connection connection = null; // 连接Channel channel = null; // 通道try {connection = factory.newConnection();// 4、获取通道channel = connection.createChannel();// TODO 开启消息确认机制channel.confirmSelect();// 5、声明 Exchange，如果不存在，则会创建String exchangeName = "exchange_confirm_2023";channel.exchangeDeclare(exchangeName, "direct");// TODO 一定要先调用监听接口，在发送消息channel.addConfirmListener(new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {System.out.println("RabbitMQ接收成功啦.....消息的标识deliveryTag=" + deliveryTag + ",批量发送多条消息multiple=" + multiple);}}, new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {System.out.println("RabbitMQ接收失败啦.....");}});for (int i = 0; i < 10; i++) {// 6、发送消息String message = "这是发布确认模式，发送的消息数据";channel.basicPublish(exchangeName, "queue_confirm_2023", null, message.getBytes());}} catch (IOException e) {System.out.println("消息发送失败啦");} catch (Exception e) {e.printStackTrace();} finally {if (null != channel) {try {channel.close();} catch (Exception e) {}}if (null != connection) {try {connection.close();} catch (Exception e) {}}}}
}

到此，RabbitMQ消息队列中的发布确认模式就介绍完啦。

综上，这篇文章结束了，主要接收消息队列RabbitMQ七种模式之Publisher Confirms发布确认模式。