【RabbitMQ】重试机制、TTL

ops/2024/9/20 10:23:50/

重试机制

在消息从Broker到消费者的传递过程中,可能会遇到各种问题,如网络故障、服务不可用、资源不足等,这些问题都可能导致消息处理失败。为了解决这些问题,RabbitMQ提供了重试机制,允许消息在处理失败之后重新发送。

但如果是程序逻辑引起的错误,那么多次重试也是不起作用的,因此设置了重试次数。

消费者确认机制为AUTO时

当消费者确认机制是AUTO时,如果程序逻辑错误,那么就会不断重试,造成消息积压。因此我们就需要设置重试次数,当多次重试还是失败,消息就会被自动确认,自然消息就会丢失。

spring:rabbitmq:host: 43.138.108.125port: 5672username: adminpassword: adminvirtual-host: mq-springboot-testlistener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 5000ms # 初始失败等待时长max-attempts: 5 # 最大重试次数
@Configuration
public class RetryConfig {@Bean("retryQueue")public Queue retryQueue() {return QueueBuilder.durable(Constants.RETRY_QUEUE).build();}@Bean("retryExchange")public Exchange retryExchange() {return ExchangeBuilder.directExchange(Constants.RETRY_EXCHANGE).durable(true).build();}@Bean("retryQueueBind")public Binding retryQueueBind(@Qualifier("retryExchange") Exchange exchange,@Qualifier("retryQueue") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("retry").noargs();}}
@RestController
@RequestMapping("/retry")
public class RetryController {@Resourcepublic RabbitTemplate rabbitTemplate;@RequestMappingpublic void retryQueue() {this.rabbitTemplate.convertAndSend(Constants.RETRY_EXCHANGE, "retry", "hello 重试机制");System.out.println("重试机制生产者发送消息成功");}}
@Configuration
public class RetryListener {@RabbitListener(queues = Constants.RETRY_QUEUE)public void retryListener(String msg) {System.out.println("获取到消息:" + msg);int n = 3 / 0;}}

上述代码和可靠性传输一文的消费者确认机制中策略为AUTO的代码类似,只不过在此配置文件中加了一个重试机制。当启动程序之后,可以看到如下结果:

5a184519433e4cdbb4e9104af3dfc725.png

重试时:

d489014923f34f40b0df4b4c9c1bea83.png 

重试结束之后: 

06595feb6ba149afa9b9477d21e456cb.png 

 从测试结果可以看出,当消费者确认机制的策略为AUTO时,遇到异常就会进行重试,当重试结束之后依然没有接收时,就会自动确认消息。

消费者确认机制为MANAUL时

当消费者确认机制是MANUL时,修改消费者代码,并启动程序,查看结果:

spring:rabbitmq:host: 43.138.108.125port: 5672username: adminpassword: adminvirtual-host: mq-springboot-testlistener:simple:acknowledge-mode: manual # 消息确认机制,手动确认retry:enabled: true # 开启消费者失败重试initial-interval: 5000ms # 初始失败等待时长max-attempts: 5 # 最大重试次数
@Configuration
public class RetryListener {@RabbitListener(queues = Constants.RETRY_QUEUE)public void retryListener(Message msg, Channel channel) throws IOException {try {System.out.println("接收到消息:" + msg);int num = 3 / 0; // 模拟处理失败channel.basicAck(msg.getMessageProperties().getDeliveryTag(), true);} catch (Exception e) {channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);}}}

a12c8ed185c8460899ded1f1709ee811.png

从测试结果可以得出,消费者确认机制为手动确认时,并不会依据配置文件中的重试次数等内容来做,而是依据消息者自身的代码实现来做实现机制。原因是因为手动确认模式下,消费者需要显示地对消息进行确认,如果消费者在消息处理过程中遇到异常,可以选择确认不确定消息,也可以选择重新入队。所以重试的控制权并不在应用程序本身,而在于代码逻辑本身。 

1. 消费者确认机制为AUTO时,如果程序逻辑异常,多次重试还是失败。那么消息就会自动确认,进而消息就会丢失。

2. 消费者确认机制为MANAUL时,如果程序逻辑异常,多次重试依然处理失败,无法被确认,消息就会积压。

3. 消费者确认机制为NONE时,不管发生什么情况,当消息从Broker内部发出时,就会自动确认,因此它不存在任何内容。

TTL

TTL,过期时间。当消息到达过期时间之后,还没有被消息,消息就会被自动清除。

RabbitMQ可以对队列和消息设置过期时间。如果两种方法同时使用,那么就以两者较小的值为准。

设置消息的TTL

@Configuration
public class TllConfig {@Bean("ttlQueue")public Queue ttlQueue() {return QueueBuilder.durable(Constants.TTL_QUEUE).build();}@Bean("ttlExchange")public Exchange ttlExchange() {return ExchangeBuilder.directExchange(Constants.TTL_EXCHANGE).build();}@Bean("ttlQueueBind")public Binding ttlQueueBind(@Qualifier("ttlExchange") Exchange exchange,@Qualifier("ttlQueue") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("ttl").noargs();}}
@RestController
@RequestMapping("/ttl")
public class TtlController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMappingpublic void ttlQueue() {MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setExpiration("50000");return message;}};this.rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, "ttl", "hello ttl", messagePostProcessor);}}

 在TTL的测试中,不需要消费者的存在,否则看不到消息在队列中的自动丢失。

设置队列的TTL

注意,设置队列的TTL,并不是过期之后删除整个队列,也是关于消息设置的。只不过投递到该消息队列的所有消息都有一个共同的过期时间而已。

@Configuration
public class TllConfig {@Bean("ttlQueue")public Queue ttlQueue() {return QueueBuilder.durable(Constants.TTL_QUEUE).ttl(5000).build();}@Bean("ttlExchange")public Exchange ttlExchange() {return ExchangeBuilder.directExchange(Constants.TTL_EXCHANGE).build();}@Bean("ttlQueueBind")public Binding ttlQueueBind(@Qualifier("ttlExchange") Exchange exchange,@Qualifier("ttlQueue") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("ttl").noargs();}}

设置队列的TTL,只需要在声明队列时给出过期时间即可。在测试的过程中,如果是先测试了消息的过期时间,那么在测试队列时,需要先将持久化的队列给删除,再启动程序。

当启动程序之后,可以看到队列上加了一个TTL的标识,表明队列的过期时间设置成功:

16dfa1734e5e43c88a5b678d594a20b1.png 

区别

设置队列的过期时间,一旦消息过期,就会从队列中删除。

设置消息的过期时间,即使消息过期,如果消息不在队首,还得等到消息到达队首之后才会进行判定是否过期。如果过期,那就删除,反之就投递到相应的消费者中。

为什么这两种方法处理的方式不一样?

因为设置队列的过期时间,那么队列中过期的消息一定在队首,RabbitMQ只需要定期从队首扫描消息是否有过期的消息即可。

而设置消息的过期时间,每条消息的过期时间都不一致,如果要删除队列的所有过期消息那么就要扫描整个队列,所以不如等到消息要进行投递时再判断消息是否过期,这样可以减少一定的资源消耗。

 


http://www.ppmy.cn/ops/113350.html

相关文章

python 安装库pycrypto失败的一系列问题[已解决]

需要安装pycrypto这个库。 /*****************安装pycryptodome***********************/说来难受,后面发现这个貌似不再维护了,可以安装另外一个库pycryptodome,是这个库的延伸版本,和这个库的作用是一样的,我也是看…

【线性回归模型】

线性回归模型 创建一些带标签的数据集𝐷 {(𝒙1, 𝑦1) , (𝒙2, 𝑦2 ), … , (𝒙𝑚, 𝑦𝑚) } x为特征,映射到对应的标签y,再引入偏置b 线性回归模…

PostgreSQL 容器安装

使用Docker安装PostgreSQL(通常简称为PgSQL)容器的步骤相对直接且简单。以下是一个详细的步骤指南,帮助你通过Docker安装并运行PostgreSQL容器: 1. 安装Docker 首先,确保系统上已经安装了Docker。可以通过访问Docker…

34.贪心算法1

0.贪心算法 1.柠檬水找零(easy) . - 力扣(LeetCode) 题目解析 算法原理 代码 class Solution {public boolean lemonadeChange(int[] bills) {int five 0, ten 0;for (int x : bills) {if (x 5) // 5 元:直接收下…

第一章 初识SpringBoot

目录 一、概述 二、原理初探 三、构建一个简单的SpringBoot应用 四、附带知识(yaml几种语法) 一、概述 Spring Boot是由Pivotal团队提供的全新框架,其设计目的是用来简化Spring应用开发和项目搭建过程。约定大于配置,化繁为简…

Arthas 全攻略:让调试变得简单

文章目录 一、简介二、命令列表 一、简介 注意 : 我安装的版本是:Arthas V3.7.2 官网:https://arthas.aliyun.com/doc/ 相关错误解决方案请看GitHub:https://github.com/alibaba/arthas/issues Alibaba开源的Java诊断工具。 从…

物流管理系统小程序的设计

管理员账户功能包括:系统首页,个人中心,用户管理,员工管理,部门管理,物品分类管理,物流公司管理,物流信息管理,配送信息管理 微信端账号功能包括:系统首页&a…

RabbitMQ 07 另两种集群方式 warren(主备模式),shovel(远程模式)

01.之前的集群有一个缺点,就是故障恢复的时候,停留在队列中的消息怎么办? 02.镜像集群模式,同步所有消息,当当前主节点不可用的时候,可以选举一个从节点来作为主节点。这样可以避免因为主节点不可用的情况…