RabbitMQ延时队列+延迟重试

RabbitMQ之TTL(JAVA设置TTL)

大白话讲解RabbitMQ消息可靠传输保障 消息确认机制 死信队列

深入分析RabbitMQ消息异常处理,及延迟队列在缓存架构中的应用

用RabbitMQ延迟队列取代定时任务

延迟队列

顾名思义,延迟队列是指被延迟消费的队列。而一般的队列,消息一旦入队了之后就会被消费者马上消费。

RabbitMQ延迟队列,主要是借助消息的TTL(Time to Live)和死信exchange(Dead Letter Exchanges:DLX)来实现。

应用场景分析

1)延迟消费

在订单系统中,一个用户下单之后通常有30分钟的时间进行支付,如果30分钟之内没有支付成功,那么这个订单将进行异常处理(如:关闭订单),这时就可以使用延迟队列来处理这些订单

还有其它一些常用的应用场景,如:物联网中的延时处理、新注册用户的活跃度调查……

2)延迟重试

主要是处理一些异常信息,如:发送失败、消费失败,可能当时由于网络抖动的问题,暂时无法访问,需要延迟再进行重试(重新发送,重新消费)

如果不使用延迟队列,那么我们只能采用定时任务,轮训数据库,方法简单好用,但性能底下,在高并发情况下容易弄死数据库间隔时间不好设置,时间过大,影响精度,过小影响性能,而且做不到按超时的时间顺序处理。

既然延迟队列这么好,我们怎样来实现延迟队列呢,本文将深入讲解延迟队列的实现原理,并实战演练延迟队列的应用

  

核心概念

1)死信队列DLX

DLX,Dead Letter Exchange 的缩写,又死信邮箱、死信交换机。DLX就是一个普通的交换机,和一般的交换机没有任何区别。

【当消息在一个队列中变成死信(dead message)时,通过这个交换机将死信发送到死信队列中】(指定好相关参数,rabbitmq会自动发送)。

什么是死信呢?什么样的消息会变成死信呢?

消息被拒绝(basic.reject或basic.nack)并且requeue=false.

消息TTL过期

队列达到最大长度(队列满了,无法再添加数据到mq中)

应用场景分析:

在定义业务队列的时候,可以考虑指定一个死信交换机,并绑定一个死信队列,当消息变成死信时,该消息就会被发送到该死信队列上,这样就方便我们查看消息失败的原因了

如何使用死信交换机呢?

定义业务(普通)队列的时候指定参数:

x-dead-letter-exchange: 用来设置死信后发送的交换机

x-dead-letter-routing-key:用来设置死信的routingKey

2)消息TTL/队列TTL(Time To Live 过期时间)

概述

消息TTL/队列TTL(Time To Live 过期时间)

队列TTL通过队列的属性来设置TTL队列中的所有消息都有相同的过期时间

消息TTL对消息进行单独设置,每条消息的TTL可以设置不同的过期时间

两者同时使用:以较小的TTL为准

消息在队列中的生存时间一旦超过设置的TTL,就会变成"死信"(Dead Message),如果绑定了死信队列,这些死信变会进入死信队列

消息在队列的生存时间一旦超过设置的TTL值,就称为dead message, 消费者将无法再收到该消息。

队列TTL与消息TTL的区别:

队列TTL:

一旦消息过期,就会立即从队列中抹去(因为过期的消息肯定处于队列的头部)

消息TTL:

即使消息过期,也不会马上从队列中抹去,因为每条消息是否过期是在即将投递到消费者之前判定的

为什么两者得处理方法不一致?

因为第一种方法里,队列中已过期的消息肯定在队列头部,RabbitMQ只要定期从队头开始扫描是否有过期消息即可,而第二种方法里,每条消息的过期时间不同,如果要删除所有过期消息,势必要扫描整个队列,所以不如等到此消息即将被消费时再判定是否过期,如果过期,再进行删除。

Queue TTL

queue.declare 命令中的 x-expires 参数控制 queue 被自动删除前可以处于未使用状态的时间。未使用的意思是 queue 上没有任何 consumer ,queue 没有被重新声明,并且在过期时间段内未调用过 basic.get 命令。该方式可用于,例如,RPC-style 的回复 queue, 其中许多 queue 会被创建出来,但是却从未被使用。

服务器会确保在过期时间到达后 queue 被删除,但是不保证删除的动作有多么的及时。在服务器重启后,持久化的 queue 的超时时间将重新计算。

用于表示超期时间的 x-expires 参数值以毫秒为单位,并且服从和 x-message-ttl 一样的约束条件,且不能设置为 0 。所以,如果该参数设置为 1000 ,则表示该 queue 如果在 1s之内未被使用则会被删除。

下面的 Java 示例创建了一个 queue ,其会在 30 分钟不使用的情况下判定为超时。

Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-expires", 1800000);  
channel.queueDeclare("myqueue", false, false, false, args);

  

总结:

死信队列 + 消息TTL = 延迟队列

1)延迟消费

 
RabbitMQ延时队列+延迟重试
 

生产者发送消息到Queue_10s

设置Queue_10s的队列TTL为10s

当消息过期后,经过DLX,进入Dead_Queue

消费者消费Dead_Queue中的消息

2)延迟重试

 
RabbitMQ延时队列+延迟重试
 

生产者发送消息到普通队列,消费者正常消费队列

生产者发送失败,发送消息到Queue_20s,消息过期后,经过DLX,进入Dead_Queue,最为数据流重新发送

消费者消费失败,发送消息到Queue_20s,消息过期后,经过DLX,进入Queue,消费者正常消费

相关推荐