devbox
Guff_9hys
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：Guff_9hys | 2024-07-08 11:13:02

赞

踩

publisher confirm机制

这篇文章，主要接收消息队列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）引入依赖

  1. <!-- 引入 RabbitMQ 依赖 -->
  2. <dependency>
  3.     <groupId>com.rabbitmq</groupId>
  4.     <artifactId>amqp-client</artifactId>
  5.     <version>5.16.0</version>
  6. </dependency>

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

  • 生产者发送消息的时候，需要调用【confirmSelect()】方法开启消息确认机制。
  • 生产者将消息发送完成之后，需要调用【waitForConfirms()】方法，阻塞等待RabbitMQ消息队列返回ACK标识。这个方法返回一个boolean类型，true表示RabbitMQ接收消息成功，false表示接收失败。
  • 【waitForConfirms()】方法还可以指定一个超时时间，如果在这个超时时间里面RabbitMQ还没有返回ACK标识，那么该方法将抛出一个InterruptedException中断异常。
  1. package com.rabbitmq.demo.confirm;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. /**
  8.  * @version 1.0.0
  9.  * @Date: 2023/2/25 16:23
  10.  * @Copyright (C) ZhuYouBin
  11.  * @Description: 消息生产者
  12.  */
  13. public class Producer {
  14.     public static void main(String[] args) {
  15.         // 1、创建连接工厂
  16.         ConnectionFactory factory = new ConnectionFactory();
  17.         // 2、设置连接的 RabbitMQ 服务地址
  18.         factory.setHost("127.0.0.1"); // 默认就是本机
  19.         factory.setPort(5672); // 默认就是 5672 端口
  20.         // 3、获取连接
  21.         Connection connection = null; // 连接
  22.         Channel channel = null; // 通道
  23.         try {
  24.             connection = factory.newConnection();
  25.             // 4、获取通道
  26.             channel = connection.createChannel();
  27.             // TODO 开启消息确认机制
  28.             channel.confirmSelect();
  29.             // 5、声明 Exchange，如果不存在，则会创建
  30.             String exchangeName = "exchange_direct_2023";
  31.             channel.exchangeDeclare(exchangeName, "direct");
  32.             // 6、发送消息
  33.             for (int i = 0; i < 10; i++) {
  34.                 // 路由键唯一标识
  35.                 String routingKey = "error";
  36.                 if (i % 3 == 0) {
  37.                     routingKey = "info";
  38.                 } else if (i % 3 == 1) {
  39.                     routingKey = "warn";
  40.                 }
  41.                 String message = "这是发布确认模式，发送的第【" + (i+1) + "】条【" + routingKey + "】消息数据";
  42.                 channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
  43.                 // 等待RabbitMQ返回ACK标识
  44.                 boolean wait = channel.waitForConfirms();
  45.                 System.out.println("RabbitMQ是否接收成功: " + wait);
  46.                 if (!wait) {
  47.                     // 消息发送失败，则可以重新发送
  48.                     channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
  49.                 }
  50.             }
  51.         } catch (Exception e) {
  52.             e.printStackTrace();
  53.         } finally {
  54.             if (null != channel) {
  55.                 try {
  56.                     channel.close();
  57.                 } catch (Exception e) {}
  58.             }
  59.             if (null != connection) {
  60.                 try {
  61.                     connection.close();
  62.                 } catch (Exception e) {}
  63.             }
  64.         }
  65.     }
  66. }

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

  • 前一种方式，是一条消息就调用一次【waitForConfirms()】方法，阻塞等待RabbitMQ的ACK确认标识。
  • 但是这种方式是非常耗时的，当需要发送的消息非常多的时候，会严重影响系统性能，所以为了解决这个问题，提出了批量确认的方法。
  • 批量确认调用【waitForConfirmsOrDie()】方法，此时会等待一批消息的ACK确认标识，如果这一批消息中存在一个消息没有被RabbitMQ成功接收，此时该方法将抛出一个【IOException】异常。
  • 所以，可以通过捕获IOException异常来判断消息是否发送成功。
  • 这种方式的缺点：当一批消息出现失败的情况时候，我们没办法知道是哪一条消息失败了，只能够重新将这一批消息重新发送。
  1. package com.rabbitmq.demo.confirm;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. import java.io.IOException;
  8.  
  9. /**
  10.  * @version 1.0.0
  11.  * @Date: 2023/2/25 16:23
  12.  * @Copyright (C) ZhuYouBin
  13.  * @Description: 消息生产者
  14.  */
  15. public class ProducerBatch {
  16.     public static void main(String[] args) {
  17.         // 1、创建连接工厂
  18.         ConnectionFactory factory = new ConnectionFactory();
  19.         // 2、设置连接的 RabbitMQ 服务地址
  20.         factory.setHost("127.0.0.1"); // 默认就是本机
  21.         factory.setPort(5672); // 默认就是 5672 端口
  22.         // 3、获取连接
  23.         Connection connection = null; // 连接
  24.         Channel channel = null; // 通道
  25.         try {
  26.             connection = factory.newConnection();
  27.             // 4、获取通道
  28.             channel = connection.createChannel();
  29.             // TODO 开启消息确认机制
  30.             channel.confirmSelect();
  31.             // 5、声明 Exchange，如果不存在，则会创建
  32.             String exchangeName = "exchange_direct_2023";
  33.             channel.exchangeDeclare(exchangeName, "direct");
  34.             // 6、发送消息
  35.             int batchSize = 3;
  36.             int count = 0;
  37.             for (int i = 0; i < 10; i++) {
  38.                 // 路由键唯一标识
  39.                 String routingKey = "error";
  40.                 if (i % 3 == 0) {
  41.                     routingKey = "info";
  42.                 } else if (i % 3 == 1) {
  43.                     routingKey = "warn";
  44.                 }
  45.                 String message = "这是发布确认模式，发送的第【" + (i+1) + "】条【" + routingKey + "】消息数据";
  46.                 channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
  47.                 // 批量确认
  48.                 if (count == batchSize) {
  49.                     // 等待RabbitMQ返回ACK标识
  50.                     channel.waitForConfirmsOrDie();
  51.                     count = 0;
  52.                 }
  53.                 count++;
  54.             }
  55.         } catch (IOException e) {
  56.             System.out.println("消息发送失败啦");
  57.         } catch (Exception e) {
  58.             e.printStackTrace();
  59.         } finally {
  60.             if (null != channel) {
  61.                 try {
  62.                     channel.close();
  63.                 } catch (Exception e) {}
  64.             }
  65.             if (null != connection) {
  66.                 try {
  67.                     connection.close();
  68.                 } catch (Exception e) {}
  69.             }
  70.         }
  71.     }
  72. }

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

  • 异步确认在消息发送之后，调用【addConfirmListener()】方法，该方法介绍两个参数，第一个参数是成功接收到ACK标识的回调方法，第二个参数是失败接收到NACK标识的回调方法。
  • 注意：一定要先调用【addConfirmListener()】监听方法，然后再发送消息，如果两者顺序反了，则监听方法不生效。
  1. package com.rabbitmq.demo.confirm;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.ConfirmCallback;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. import java.io.IOException;
  9.  
  10. /**
  11.  * @version 1.0.0
  12.  * @Date: 2023/2/25 16:23
  13.  * @Copyright (C) ZhuYouBin
  14.  * @Description: 消息生产者
  15.  */
  16. public class ProducerAsync {
  17.     public static void main(String[] args) {
  18.         // 1、创建连接工厂
  19.         ConnectionFactory factory = new ConnectionFactory();
  20.         // 2、设置连接的 RabbitMQ 服务地址
  21.         factory.setHost("127.0.0.1"); // 默认就是本机
  22.         factory.setPort(5672); // 默认就是 5672 端口
  23.         // 3、获取连接
  24.         Connection connection = null; // 连接
  25.         Channel channel = null; // 通道
  26.         try {
  27.             connection = factory.newConnection();
  28.             // 4、获取通道
  29.             channel = connection.createChannel();
  30.             // TODO 开启消息确认机制
  31.             channel.confirmSelect();
  32.             // 5、声明 Exchange，如果不存在，则会创建
  33.             String exchangeName = "exchange_confirm_2023";
  34.             channel.exchangeDeclare(exchangeName, "direct");
  35.             // TODO 一定要先调用监听接口，在发送消息
  36.             channel.addConfirmListener(new ConfirmCallback() {
  37.                 @Override
  38.                 public void handle(long deliveryTag, boolean multiple) throws IOException {
  39.                     System.out.println("RabbitMQ接收成功啦.....消息的标识deliveryTag=" + deliveryTag 
  40.                             + ",批量发送多条消息multiple=" + multiple);
  41.                 }
  42.             }, new ConfirmCallback() {
  43.                 @Override
  44.                 public void handle(long deliveryTag, boolean multiple) throws IOException {
  45.                     System.out.println("RabbitMQ接收失败啦.....");
  46.                 }
  47.             });
  48.             for (int i = 0; i < 10; i++) {
  49.                 // 6、发送消息
  50.                 String message = "这是发布确认模式，发送的消息数据";
  51.                 channel.basicPublish(exchangeName, "queue_confirm_2023", null, message.getBytes());
  52.             }
  53.         } catch (IOException e) {
  54.             System.out.println("消息发送失败啦");
  55.         } catch (Exception e) {
  56.             e.printStackTrace();
  57.         } finally {
  58.             if (null != channel) {
  59.                 try {
  60.                     channel.close();
  61.                 } catch (Exception e) {}
  62.             }
  63.             if (null != connection) {
  64.                 try {
  65.                     connection.close();
  66.                 } catch (Exception e) {}
  67.             }
  68.         }
  69.     }
  70. }

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

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

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Guff_9hys/article/detail/798554
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号