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）延迟消费

 

 

生产者发送消息到Queue_10s

设置Queue_10s的队列TTL为10s

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

消费者消费Dead_Queue中的消息

 

2）延迟重试

 

 

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

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

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