当前位置: 首页 > java >正文

RabbitMQ——消息确认

一、消息确认机制

生产者发送的消息,可能有以下两种情况:

1> 消息消费成功

2> 消息消费失败

为了保证消息可靠的到达消费者(!!!注意:消息确认机制和前面的工作模式中的publisher confirms模式有很大区别,消息确认保证的是消息可靠的到达消费者,而publisher confirms保证的是消息可靠的到达RabbitMQServer),RabbitMQ引入了消息确认机制:

消费者在消费消息时,可以指定autoAck参数,对应着两种确认方式

(1)自动确认:消息只要到达消费者就会自动确认,不会考虑消费者是否正确消费了这些消息,直接从 内存/磁盘 中删除消息;

(2)手动确认:消息到达消费者,不会自动确认,会等待消费者调用Basic.Ack命令,才会从内存/磁盘 移除这条消息。

DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, 
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息: " + new String(body));}
};
channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, true, consumer);

如果将basicConsume中的true改为false,对于RabbitMQ服务器来说,消息分成两个部分:

1>  未发送给消费者的消息;

2>  已经发送给消费者,但还没有被确认的消息。

对应管理界面:


二、手动确认方法

RabbitMQ提供了两种手动确认,分别是肯定确认和否定确认,对应着3个方法:

(1)肯定确认:Channel.basicAck(long deliveryTag, boolean multiple)

    其中,deliveryTag为消息的唯一表示,在每一个channel中都是唯一的,相当于TCP协议中的序号的作用,用来确认哪条消息已经收到,multiple为true表示是否批量确认,同样与TCP中确认序号的作用类似,表示在这个deliveryTag前的消息都已收到,为false则表示一次只确认一条消息。

(2)否定确认: Channel.basicReject(long deliveryTag, boolean requeue)

   如果消息到达消费者后,消费者未正确处理这条消息(如发生异常),就可以通过这个方法进行否定确认,其中,参数requeue表示这条消息是否需要重新入队,这个方法一次只能确认一条消息。

(3)否定确认:Channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)

   与basicAck一样,这个方法可以通过multiple来实现批量确认


三、使用Spring Boot演示消息确认机制

  演示之前,需要先了解Spring-AMQP的确认机制,它和RabbitMQ JDK Client库的确认机制有些许不同:

1. AcknowledgeMode.NONE

  和RabbitMQ JDK Client库的自动确认机制一样,只要消息到达消费者,这条消息就会从队列中移除。

2. AcknowledgeMode.AUTO(和JDK Client库的区别)

  消息到达消费者且处理成功,才会自动确认,如果消息在处理过程中发生了异常,则不会自动确认。

3. AcknowledgeMode.MANUAL

  和JDK Client库一样,属于手动确认机制,同样分为肯定确认和否定确认。

接下来,进行准备工作,创建一个Spring Boot项目,添加RabbitMQ依赖:

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

添加RabbitMQ相关配置:

spring:rabbitmq:addresses: amqp://study:study@110.41.17.130:5672/extensionlistener:simple:acknowledge-mode: none #表示当前确认机制未AcknowledgeMode.NONE

添加下图的包,创建对应的类:

 


3.1  AcknowledgeMode.NONE

(1)编写常量类(由于只是验证消息确认机制,可以随便使用一种工作模式)

public class Constants {public static final String ACK_QUEUE = "ack.queue";public static final String ACK_EXCHANGE = "ack.exchange";
}

(2)配置声明队列、交换机、交换机与队列绑定关系

@Configuration
public class RabbitMQConfig {//1.声明队列@Bean("ackQueue")public Queue ackQueue(){return QueueBuilder.durable(Constants.ACK_QUEUE).build();}//2.声明交换机@Bean("ackExchange")public FanoutExchange ackExchange(){return ExchangeBuilder.fanoutExchange(Constants.ACK_EXCHANGE).build();}//3.声明队列与交换机的绑定关系@Beanpublic Binding ackBinding(@Qualifier("ackExchange") FanoutExchange fanoutExchange,@Qualifier("ackQueue") Queue queue){return BindingBuilder.bind(queue).to(fanoutExchange);}
}

(3)编写生产者代码

@RestController
@RequestMapping("/producer")
public class ProducerController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/ack")public String ack(){rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,"","consumer ack mode test...");return "发送消息成功";}
}

(4)编写消费者代码

@Component
public class AckListener {@RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {long deliveryTag = message.getMessageProperties().getDeliveryTag();System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());//业务逻辑处理System.out.println("业务逻辑处理");
//            int num = 10/0;System.out.println("业务处理完成");}
}

(5)测试正常消费消息的情况


(6)测试消息消费异常情况(将消费者代码中得 int num  = 10/0  注解打开)


(7)总结

  经过确认,可以发现AcknowledgeMode.NONE机制,无论消费者是否正确消费消息,都会自动确认,不会保留异常消费的消息


3.2 AcknowledgeMode.AUTO

(1) 修改配置文件中的 acknowledge-mode: none 为 acknowledge-mode: auto

(2)演示消息正常消费的情况(将 int num = 10/0 注释)

  @RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {long deliveryTag = message.getMessageProperties().getDeliveryTag();System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());//业务逻辑处理System.out.println("业务逻辑处理");
//            int num = 10/0;System.out.println("业务处理完成");}

运行程序,访问 producer/ack 接口发送消息:


(3)演示消息处理异常情况(取消 int num = 10/0 的注释)

 @RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {long deliveryTag = message.getMessageProperties().getDeliveryTag();System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());//业务逻辑处理System.out.println("业务逻辑处理");int num = 10/0;System.out.println("业务处理完成");}

运行程序,访问生产者端口:

消息没有成功消费,如果此时我们修改代码(将 int num = 10/0 注释,使程序不再发生异常),再次运行程序,队列中保存的消息就会被正常消费:


3.3 AcknowledgeMode.MANUAL

  (1) 修改配置文件中 acknowledge-mode: auto 为 acknowledge-mode: manul

(2)修改消费者代码(由于是手动确认,需要在代码中添加正常消费时的 “肯定确认” 和 消费异常时的 “否定确认”)

    @RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try{System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());//业务逻辑处理System.out.println("业务逻辑处理");
//                int num = 10/0;System.out.println("业务处理完成");//消息正常消费,肯定确认channel.basicAck(deliveryTag,false);//单条确认}catch (Exception e){//消息消费异常,否定确认channel.basicNack(deliveryTag,false,true);//单条否定,异常后重新入队}}

(3)消息正常消费时的情况(注释 int num = 10/0)

接下来注释掉手动肯定确认的这行代码,看看会发生什么情况:

 //消息正常消费,肯定确认//channel.basicAck(deliveryTag,false);//单条确认

再次运行程序并通过接口 producer/ack 发送消息:

可以看到,在AcknowledgeMode.MANUAL机制下,如果不手动确认,队列不会移除已经被正常消费的消息:


(3)消息消费异常的情况下(取消 int num = 10/0 的注释)

接下来注释掉channel.basicNack,运行程序,访问接口发送消息:

如果将参数requeue置为false,会怎么样?

channel.basicNack(deliveryTag,false,false);//单条否定,重新发送消息

运行程序:


(4)总结

1> 无论是否正常处理消息都要进行手动确认;

2> 正常处理消息但未手动确认,管理界面中的队列会有 一条/多条 Unacked 的消息(重新启动程序后会重新消费);

3> 异常处理消息且未手动确认,也会有 一条/多条 Unacked 的消息(重启程序同样重新消费);

4> 如果异常处理,将requeue置为false,队列不会保存 这条/多条 异常消费的消息

                                               置为true,队列会不断重新发送 这条/多条 消息

http://www.xdnf.cn/news/7778.html

相关文章:

  • NLP学习路线图(三): 微积分(梯度、导数等)
  • 有没有其他影视app可以像群晖video station一样可以被Windows的本地网络驱动器找到
  • 【Vue3】数据的返回和响应式处理(ref reactive)
  • 自建srs实时视频服务器支持RTMP推流和拉流
  • Kotlin 极简小抄 P8(不可空类型、可空类型、注意事项、非空断言 !!)
  • 什么是endpoints?
  • php://filter的trick
  • Server-Driven UI:Kotlin 如何重塑动态化 Android 应用开发
  • 《算法笔记》12.1小节——字符串专题->字符串hash进阶 问题 A: 求最长公共子串(串)
  • 代码随想录打卡|Day45 图论(孤岛的总面积 、沉没孤岛、水流问题、建造最大岛屿)
  • ARM反汇编浅析
  • 【JAVA】比较器Comparator与自然排序(28)
  • Flannel后端为UDP模式下,分析数据包的发送方式(一)
  • 【react18】在styled-components中引入图片报错
  • 项目中Warmup耗时高该如何操作处理
  • 深度解析 Java 中介者模式:重构复杂交互场景的优雅方案
  • 详解 C# 中基于发布-订阅模式的 Messenger 消息传递机制:Messenger.Default.Send/Register
  • 服务器网络配置 netplan一个网口配置两个ip(双ip、辅助ip、别名IP别名)
  • Java详解LeetCode 热题 100(18):LeetCode 73. 矩阵置零(Set Matrix Zeroes)详解
  • 广州卓远VR受邀参加2025智能体育典型案例调研活动,并入驻国体华为运动健康联合实验室!
  • 深入解析异步编程:Java NIO、Python `async/await` 与 C# `async/await` 的对比
  • junit单元测试
  • Ajax研究
  • [Linux] Linux信号量深度解析与实践(代码示例)
  • VLA模型:自动驾驶与机器人行业的革命性跃迁,端到端智能如何重塑未来?
  • docker 启动一个python环境的项目
  • 零数组变换 二分+查分数组||线段树lazy
  • 算法C++最大公约数
  • Linux条件变量
  • 从零基础到最佳实践:Vue.js 系列(4/10):《Vue Router 路由管理:深入探索与实战应用》