1、延迟队列
延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。
在电商系统中，当用户提交订单超过 30 分钟未支付就是自动取消。一般的解决方案是使用定时任务来轮训数据库，然后找到过期的订单来修改订单状态。当系统的数据量小的时候下没有什么问题。但是如果数据量一大这种方式就会特别消耗资源。并且这种方式对于订单的取消还有延迟。
而使用 RabbitMQ 的延迟队列就是解决这些问题的。
使用RabbitMQ来实现延迟队列必须先了解RabbitMQ的两个概念：消息的 TTL 和死信 Exchange，通过这两者的组合来实现上述需求。
2、RabbitMQ 实现迟队列
AMQP协议和 RabbitMQ 队列本身没有直接支持延迟队列功能，但是可以通过以下 消息的TTL 与 死信 Exchange 这两个特性特性模拟出延迟队列的功能。
2.1 Time To Live(TTL)
消息的 TTL 就是消息的存活时间。RabbitMQ 可以对队列和消息分别设置TTL。RabbitMQ 可以针对 Queue 设置 x-expires 或者 针对 Message 设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)。超过了这个时间，我们认为这个消息就死了，称之为死信。如果队列设置了，消息也设置了，那么会取小的。
所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。
RabbitMQ针对队列中的消息过期时间有两种方法可以设置。
- A: 通过队列属性设置，队列中所有消息都有相同的过期时间。
- B: 对消息进行单独设置，每条消息TTL可以不同。
如果同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead letter
2.2 Dead Letter Exchanges（DLX）
RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。
- x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
- x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送
队列满足以下条件就会出现 dead letter 情况有：
- 消息或者队列的TTL过期
- 队列达到最大长度
- 消息被消费端拒绝（basic.reject or basic.nack）并且requeue = false
综合上述两个特性，设置了TTL规则之后当消息在一个队列中变成死信时，利用DLX特性它能被重新转发到另一个Exchange或者Routing Key，这时候消息就可以重新被消费了。
3、代码演示
这里的场景是订单创建，当订单创建成功的时候会把消息发送到 RabbitMQ 的扇形交换机里面(发布/订阅模式)。因为订单创建成功有多个消息者：一个是订单创建成功需要向会员发送优惠券，别一个是订单创建成功如果多久没有支付就会关闭订单。这里就用到了 RabbitMQ 中的死信队列。
这个项目是基于 Spring Boot，如果大家不清楚 Spring Boot 的，可以自行了解一下。
3.1 项目依赖
下面是定义在 xml 里面的项目依赖 Jar 包管理，使用的是 maven 进行项目管理。
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.1.RELEASE</version>
<relativePath/>
</parent>
<groupId>cn.carlzone.netty</groupId>
<artifactId>netty-demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>netty-demo</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
3.2 application.properties
在 Spring 配置文件中配置需要依赖的 rabbitmq 服务相关的信息。
appplication.properties
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
3.3 MQ 常量类
这个类是定义 mq 相关的常量信息。
MQConstant
public abstract class MQConstant {
/** 订单创建 fanout(扇形) 交换机 */
public final static String PAY_ORDER_CREATE_FANOUT_EXCHANGE = "pay.order.create.fanout.exchange";
/** 订单创建 Dead Letter 交换机 */
public final static String PAY_ORDER_CREATE_DEAD_LETTER_EXCHANGE = "pay.order.create.dead.letter.exchange";
/** 订单创建优惠券订阅队列 */
public final static String PAY_ORDER_CREATE_COUPON_QUEUE = "pay.order.create.coupon";
/** 订单创建 6 秒不支付死信队列 (TTL QUEUE) */
public final static String PAY_ORDER_CREATE_DEAD_LETTER_QUEUE = "pay.order.create.dead.letter";
/** 死信转发队列(DLX QUEUE) */
public final static String PAY_ORDER_CREATE_EXPIRE_QUEUE = "pay.order.create.expire";
public final static String PAY_ORDER_CREATE_EXPIRE_ROUTING_KEY = "pay.order.create.expire.routing.key";
}
这个类是抽象类，在定义常量或者工具类的时候最好把这个类定义成抽象类。因为常量类或者工具类都是不需要实例化的。
3.4 Rabbit MQ Java 配置类
RabbitQueueConfiguration 是定义 RabbitMQ 里面的队列，交换机以及绑定信息。
@Configuration
public class RabbitQueueConfiguration {
@Bean
public MessageConverter jackson2JsonMessageConverter(){
return new Jackson2JsonMessageConverter();
}
/**
* 订单创建扇形交换机
* @return
*/
@Bean
public FanoutExchange orderCreateFanoutExchange(){
return new FanoutExchange(MQConstant.PAY_ORDER_CREATE_FANOUT_EXCHANGE, true, false);
}
/**
* 订单创建死信交换机
* @return
*/
@Bean
public DirectExchange orderCreateDeadLetterExchange(){
return new DirectExchange(MQConstant.PAY_ORDER_CREATE_DEAD_LETTER_EXCHANGE, true, false);
}
/**
* 订单创建 -- 优惠券消费队列
* @return
*/
@Bean
public Queue orderCreateCouponQueue(){
return new Queue(MQConstant.PAY_ORDER_CREATE_COUPON_QUEUE,true);
}
/**
* 订单创建 -- 死信队列
* @return
*/
@Bean
public Queue orderCreateDeadLetterQueue(){
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", MQConstant.PAY_ORDER_CREATE_DEAD_LETTER_EXCHANGE);
arguments.put("x-dead-letter-routing-key", MQConstant.PAY_ORDER_CREATE_EXPIRE_ROUTING_KEY);
arguments.put("x-message-ttl", 6000);
// 是否持久化
boolean durable = true;
// 仅创建者可以使用的私有队列，断开后自动删除
boolean exclusive = false;
// 当所有消费客户端连接断开后，是否自动删除队列
boolean autoDelete = false;
return new Queue(MQConstant.PAY_ORDER_CREATE_DEAD_LETTER_QUEUE,durable, exclusive, autoDelete, arguments);
}
/**
* 订单创建 -- 过期消费队列
* @return
*/
@Bean
public Queue orderCreateExpireQueue() {
Queue queue = new Queue(MQConstant.PAY_ORDER_CREATE_EXPIRE_QUEUE,true,false,false);
return queue;
}
/**
* {订单创建扇形交换机}绑定{优惠券消费队列}
* @return
*/
@Bean
public Binding orderCreateCouponBinding() {
return BindingBuilder.bind(orderCreateCouponQueue()).to(orderCreateFanoutExchange());
}
/**
* {订单创建扇形交换机}绑定{死信队列}
* @return
*/
@Bean
public Binding orderCreateDeadLetterBinding() {
return BindingBuilder.bind(orderCreateDeadLetterQueue()).to(orderCreateFanoutExchange());
}
/**
* {死信队列}绑定{订单创建死信交换机}
* @return
*/
@Bean
public Binding orderCreateDeadLetterExpireBinding() {
return BindingBuilder.bind(orderCreateDeadLetterQueue()).to(orderCreateDeadLetterExchange()).withQueueName();
}
/**
* {订单创建死信交换机}通过{死信队列中的路由键}与{过期消费队列}绑定起来
* @return
*/
@Bean
public Binding orderCreateExpireBinding() {
return BindingBuilder.bind(orderCreateExpireQueue()).to(orderCreateDeadLetterExchange()).with(MQConstant.PAY_ORDER_CREATE_EXPIRE_ROUTING_KEY);
}
}
这里配置的是 6 秒，如果订单 6 秒没有支付就会触发订单关闭动作。
3.5 发送订单创建类
这里通过 http 请求模拟订单创建成功，并发送信息到 RabbitMQ 服务器。
@Data
public class Order {
private String id;
private String name;
}
@RestController
public class MessageController {
@Resource
private RabbitTemplate rabbitTemplate;
@RequestMapping("send")
public String send(@RequestBody Order order){
rabbitTemplate.convertAndSend(MQConstant.PAY_ORDER_CREATE_FANOUT_EXCHANGE, "", order);
return "OK";
}
}
3. 6 订单创建监听器
下面就是关注订单创建的两个监听器，一个模拟优惠券发送，一个模拟订单关闭。他们的逻辑是格式化打印当前时间。
@Component
public class OrderCreateCouponListener {
@RabbitListener(queues = MQConstant.PAY_ORDER_CREATE_COUPON_QUEUE)
public void process(Order message) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String time = sdf.format(new Date());
System.out.println("订单创建发送优惠券：" + time + " 收到消息," + message);
}
}
@Component
public class OrderCreateExpireListener {
@RabbitListener(queues = MQConstant.PAY_ORDER_CREATE_EXPIRE_QUEUE)
public void process(Order message) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String time = sdf.format(new Date());
System.out.println("订单创建订单过期：" + time + " 收到消息," + message);
}
}
4、测试
我们使用 postman 发送一条 http 请求。信息为订单 ID 是 11，订单的商品名称是iphone 6s
接着我们可以在控制台打死，优惠券监听器获取到的时间与订单创建过期消费的监听器相差 6 秒。这个是符合我们预期的。
参考文章：
- https://blog.csdn.net/skiof007/article/details/80914318
转载：https://blog.csdn.net/u012410733/article/details/87922920