一、场景：
1.定时发布文章
2.秒杀之后，给30分钟时间进行支付，如果30分钟后，没有支付，订单取消。
3.预约餐厅，提前半个小时发短信通知用户。
A -> 13:00 17:00 16:30 延迟时间： 7*30 * 60 * 1000
B -> 11:00 18:00 17:30 延迟时间： 13*30 * 60 * 1000
C -> 8:00 14:00 13:30 延迟时间： 11*30 * 60 * 1000
二、3种应用方式：
1、第一种方式：创建具有超时功能且绑定死信交换机的消息队列
@Bean
public Queue directQueueLong(){
return QueueBuilder.durable("业务队列名称")
.deadLetterExchange("死信交换机名称")
.deadLetterRoutingKey("死信队列 RoutingKey")
.ttl(20000) // 消息停留时间
//.maxLength(500)
.build();
}
监听死信队列，即可处理超时的消息队列
缺点：
上述实现方式中，ttl延时队列中所有的消息超时时间都是一样的，如果不同消息想设置不一样的超时时间，就需要建立多个不同超时时间的消息队列，比较麻烦，且不利于维护。
2、第二种方式：创建通用延时消息
rabbitTemplate.convertAndSend("交换机名称", "RoutingKey","对象",
message => {
message.getMessageProperties().setExpiration(String.valueOf(5000))
return message;
}
);
缺点：
该种方式可以创建一个承载不同超时时间消息的消息队列，但是这种方式有一个问题，如果消息队列中排在前面的消息没有到超时时间，即使后面的消息到了超时时间，先到超时时间的消息也不会进入死信队列，而是先检查排在最前面的消息队列是否到了超时时间，如果到了超时时间才会继续检查后面的消息。
3、第三种方式：使用rabbitmq的延时队列插件，实现同一个队列中有多个不同超时时间的消息，并按时间超时顺序出队
1）下载延迟插件：Releases · rabbitmq/rabbitmq-delayed-message-exchange · GitHub
注意下载的插件版本要和你的MQ版本一致。
2）安装插件并启用
①、我用的是 Docker 客户端，下载完成后直接把插件放在 /root 目录，然后拷贝到容器内plugins目录下（rabbitmq是容器的name，也可以使用容器id）
docker cp /opt/rabbitmq/rabbitmq_delayed_message_exchange-3.10.0.ez rabbitmq:/plugins②、进入 Docker 容器
docker exec -it rabbitmq /bin/bash③、在plugins内启用插件
#先执行，解除防火墙限制，增加文件权限
umask 0022
rabbitmq-plugins enable rabbitmq_delayed_message_exchange④、退出容器
exit⑤、重启容器
docker restart 容器id/容器名
docker restart rabbitmq⑥、通过UI界面查看
⑦、原理解释
三、 java使用
1、消费者（需要定制一个交换机CustomExchange）
/*
* Copyright (c) 2020, 2024, All rights reserved.
*
*/
package com.by.consumer;
import cn.hutool.core.map.MapUtil;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
/**
* <p>Project: spring-boot-rabbitMQ - DirectConsumer</p>
* <p>Powered by scl On 2024-04-07 16:57:20</p>
* <p>描述：<p>
*
* @author 孙臣龙 [1846080280@qq.com]
* @version 1.0
* @since 17
*/
@Configuration
public class DelayConsumer {
//注册队列
@Bean
public Queue delayQueue() {
return QueueBuilder.durable("Delay_Q01").build();
}
//注册交换机
@Bean
public CustomExchange delayExchange() {
Map<String, Object> map = MapUtil.of("x-delayed-type", "direct");
return new CustomExchange("Delay_E01", "x-delayed-message", true, false, map);
}
//绑定交换机和队列
@Bean
public Binding binding() {
return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("RK01").noargs();
}
//启动一个消费者
@RabbitListener(queues = "Delay_Q01")
public void receiveMessage(OrderKO msg) {
System.out.println("消费者1：" + msg);
}
}2、生产者
/*
* Copyright (c) 2020, 2024, All rights reserved.
*
*/
package com.by.provider;
import com.by.consumer.OrderKO;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* <p>Project: spring-boot-rabbitMQ - DirectProvider</p>
* <p>Powered by scl On 2024-04-07 17:06:41</p>
* <p>描述：<p>
*
* @author 孙臣龙 [1846080280@qq.com]
* @version 1.0
* @since 17
*/
@Service
public class DelayProvider {
@Autowired
private RabbitTemplate rabbitTemplate;
//启动一个生产者
public void send(OrderKO orderKO) {
rabbitTemplate.convertAndSend("Delay_E01", "RK01", orderKO, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
int id = orderKO.getId();
int delay = 0;
if (id == 1) {
delay = 5000;
} else if (id==2) {
delay = 4000;
}else if (id==3) {
delay = 3000;
}else if (id==4) {
delay = 2000;
}else if (id==5) {
delay = 1000;
}
//延迟交换机设置delay参数，设置消息的延迟时间，单位毫秒
message.getMessageProperties().setDelay(delay);
return message;
}
});
}
}3、测试
/**
* 延迟交换机
* @throws InterruptedException
* @throws IOException
*/
@Test
void test5() throws InterruptedException, IOException {
for (int i = 1; i <= 5; i++) {
OrderKO orderKO = OrderKO.builder().id(i).name("孙臣龙" + i).build();
System.out.println("发送消息"+i);
delayProvider.send(orderKO);
}
Thread.sleep(10000);
//System.in.read();
}4、结果展示
