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)延迟消费
生产者发送消息到Queue_10s
设置Queue_10s的队列TTL为10s
当消息过期后,经过DLX,进入Dead_Queue
消费者消费Dead_Queue中的消息
2)延迟重试
生产者发送消息到普通队列,消费者正常消费队列
生产者发送失败,发送消息到Queue_20s,消息过期后,经过DLX,进入Dead_Queue,最为数据流重新发送
消费者消费失败,发送消息到Queue_20s,消息过期后,经过DLX,进入Queue,消费者正常消费