Springboot中使用RabbitMQ
1、准备
1.1、使用docker安装rabbitMQ
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management
端口说明：
4369，25672 Erlang发现，集群端口
5672，5671 AMQP端口
15672 web管理后台端口
61613，61614 STOMP协议端口
1883，8883 MQTT协议端口端口详细说明查看RabbitMQ端口说明
启动完毕登录web管理页面 ： http://ip:15672/ 账号密码都是 guest
1.2、创建项目
创建Springboot项目，导入RabbitMQ依赖：
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>配置文件配置连接信息
spring: rabbitmq: host: 192.168.5.37 port: 5672 #账号密码 username: guest password: guest # 虚拟主机名字 virtual-host: /启动类添加 @EnableRabbit注解
2、创建exchange、queue、binding
2.1 手动创建
@Autowired
private AmqpAdmin amqpAdmin;
/**
* 创建交换机(Exchange)、队列(Queue)、绑定(Binding)
* */
public void create(){
// 创建交换机 参数：交换机名字，是否持久化，是否自动删除，自定义参数
DirectExchange directExchange = new DirectExchange("java-exchange", true, false, null);
amqpAdmin.declareExchange(directExchange);
//创建队列 参数：队列名，是否持久化，是否排它（只能被声明的连接使用，一旦连接其它无法连接），是否自动删除，自定义参数
Queue queue = new Queue("java-queue",true,false,false,null);
amqpAdmin.declareQueue(queue);
/**
* 创建绑定关系 参数：
* String destination, 目的地 (队列名或者交换机名)
* Binding.DestinationType destinationType, 目的地类型(队列或者交换机)
* String exchange, 交换机
* String routingKey, 路由键
* @Nullable Map<String, Object> arguments 自定义参数
*/
Binding binding = new Binding("java-queue",
Binding.DestinationType.QUEUE,
"java-exchange",
"java",
null);
amqpAdmin.declareBinding(binding);
}2.2 Spring自动注入
Spring支持以配置文件的形式创建exchange、queue、binding
例：
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MQBeanConfig {
@Bean
public Queue queue1(){
return new Queue("queueName", true, false, false);
}
@Bean
public Exchange exchange1(){
return new TopicExchange("exchangeName",true,false);
}
@Bean
public Binding binding1(){
return new Binding("queueName",
Binding.DestinationType.QUEUE,
"exchangeName",
"routingKey",
null
);
}
}
注：@Bean里面的属性注入时，如果MQ中已存在也不会进行覆盖
3、收发消息
配置使用JSON序列化消息内容：RabbitConfig
@Configuration
public class RabbitConfig {
/**
* 配置使用JSON序列化消息内容
* */
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
}发送消息
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMessage(){
//发送消息 参数：交换机，路由键，消息
rabbitTemplate.convertAndSend("java-exchange","java","msg");
}接收消息
@RabbitListener ：用在方法或类上 用于接收消息，放在类上时表示监听哪些队列，接收方法使用@RabbitHandler
@RabbitHandler ： 只能用在方法上 用于重载接收不同参数的消息
多个服务监听同一个队列时，同一条消息只会被一个服务消费
/**
* queues：声明需要监听的队列名
* 方法参数可以写以下类型：
* 1、Message message：原生消息类型：消息头+消息体
* 2、T<发送消息的类型>
*/
@RabbitListener(queues = {"java-queue"})
public void receiveMessage(Message message,Object obj){
//消息体
byte[] body = message.getBody();
//消息头
MessageProperties properties = message.getMessageProperties();
}4、消息确认机制
保证消息不丢失，可靠抵达，可以使用事务消息，性能下降250倍。
4.1、消息抵达p->b
消息由product->Broker
- 配置文件添加 发送消息确认机制
spring:
rabbitmq:
host: 192.168.5.37
port: 5672
#账号密码
username: guest
password: guest
# 虚拟主机名字
virtual-host: /
#开启消息发送端（生产者）确认机制
publisher-confirms: true- 配置类配置RabbitTemplate，设置确认回调
@Configuration
public class RabbitConfig {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 配置使用JSON序列化消息内容
* */
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
/**
* 配置消息发送端确认模式
* */
@PostConstruct //构造器创建后执行此方法
public void initRabbitTemplate(){
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
// correlationData 当前消息唯一关联数据（消息id） b 消息是否成功收到 s 失败的原因
public void confirm(CorrelationData correlationData, boolean b, String s) {
}
});
}
}4.2、消息抵达b->q
- 配置抵达队列确认
spring:
rabbitmq:
host: 192.168.5.37
port: 5672
#账号密码
username: guest
password: guest
# 虚拟主机名字
virtual-host: /
#开启消息发送端（生产者）确认机制
publisher-confirms: true
#开启发送失败退回（消息有没有找到合适的队列）
publisher-returns: true
#只要抵达队列，以异步模式优先回调
template:
mandatory: true- 配置类配置RabbitTemplate，设置确认回调
/**
* 配置消息发送端确认模式
* */
@PostConstruct //构造器创建后执行此方法
public void initRabbitTemplate(){
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
// correlationData 当前消息唯一关联数据（消息id） b 消息是否成功收到 s 失败的原因
public void confirm(CorrelationData correlationData, boolean b, String s) {
}
});
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
//只要消息没有抵达指定队列就触发此回调
public void returnedMessage(ReturnedMessage returnedMessage) {
}
});
}4.3、消息消费q->c
默认是自动确认的，只要消息被接收就会自动从队列中移除
- 配置手动确认模式
spring:
rabbitmq:
host: 192.168.5.37
port: 5672
#账号密码
username: guest
password: guest
# 虚拟主机名字
virtual-host: /
#开启消息发送端（生产者）确认机制
publisher-confirms: true
#开启发送失败退回（消息有没有找到合适的队列）
publisher-returns: true
#只要抵达队列，以异步模式优先回调
template:
mandatory: true
#手动确认机制
listener:
simple:
acknowledge-mode: manual- 手动确认
@RabbitHandler
public void receiveMessage(Message message, Object obj, Channel channel){
//.....业务处理
//消息头
MessageProperties properties = message.getMessageProperties();
try {
//签收货物 参数：channel内消息顺序， 是否批量确认(false逐个确认)
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
} catch (IOException e) {
e.printStackTrace();
System.err.println("网络中断");
}
}- 拒接货物
@RabbitHandler
public void receiveMessage(Message message, Object obj, Channel channel){
//消息体
byte[] body = message.getBody();
//消息头
MessageProperties properties = message.getMessageProperties();
try {
//拒接货物 方式1 参数：channel内消息顺序， 是否批量， 是否重新入队
channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
//拒接货物 方式2 参数：channel内消息顺序， 是否重新入队
channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
} catch (IOException e) {
e.printStackTrace();
System.err.println("网络中断");
}
}5、延时队列
以下内容来源于博客：https://www.cnblogs.com/mfrank/p/11260355.html
延时队列：是用来存放需要在指定时间被处理的元素的队列。
什么时候需要用延时队列呢？
- 订单在十分钟之内未支付则自动取消。
- 新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
- 账单在一周内未支付，则自动结算。
- 用户注册成功后，如果三天内没有登陆则进行短信提醒。
- 用户发起退款，如果三天内没有得到处理则通知相关运营人员。
- 预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。
解决方法：使用RabbitMQ消息的TTL（消息存活时间）和死信Exchange结合
消息的流向：
RabbitMQ可以对队列和消息分别设置TTL：
- 对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。
- 如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间，两者是一样的效果。
消息在什么时候会变成死信：
- 消息拒绝并且没有设置重新入队
- 消息TTL过期
- 消息堆积，并且队列达到最大长度，先入队的消息会变成DL（队列满了，无法再添加）
队列设置TTL
生产者生产一条延时消息，根据需要延时时间的不同，利用不同的routingkey将消息路由到不同的延时队列，每个队列都设置了不同的TTL属性，并绑定在同一个死信交换机中，消息过期后，根据routingkey的不同，又会被路由到不同的死信队列中，消费者只需要监听对应的死信队列进行处理即可。
下面来看代码：
先声明交换机、队列以及他们的绑定关系：
@Configuration
public class RabbitMQConfig {
public static final String DELAY_EXCHANGE_NAME = "delay.queue.demo.business.exchange";
public static final String DELAY_QUEUEA_NAME = "delay.queue.demo.business.queuea";
public static final String DELAY_QUEUEB_NAME = "delay.queue.demo.business.queueb";
public static final String DELAY_QUEUEA_ROUTING_KEY = "delay.queue.demo.business.queuea.routingkey";
public static final String DELAY_QUEUEB_ROUTING_KEY = "delay.queue.demo.business.queueb.routingkey";
public static final String DEAD_LETTER_EXCHANGE = "delay.queue.demo.deadletter.exchange";
public static final String DEAD_LETTER_QUEUEA_ROUTING_KEY = "delay.queue.demo.deadletter.delay_10s.routingkey";
public static final String DEAD_LETTER_QUEUEB_ROUTING_KEY = "delay.queue.demo.deadletter.delay_60s.routingkey";
public static final String DEAD_LETTER_QUEUEA_NAME = "delay.queue.demo.deadletter.queuea";
public static final String DEAD_LETTER_QUEUEB_NAME = "delay.queue.demo.deadletter.queueb";
// 声明延时Exchange
@Bean("delayExchange")
public DirectExchange delayExchange(){
return new DirectExchange(DELAY_EXCHANGE_NAME);
}
// 声明死信Exchange
@Bean("deadLetterExchange")
public DirectExchange deadLetterExchange(){
return new DirectExchange(DEAD_LETTER_EXCHANGE);
}
// 声明延时队列A 延时10s
// 并绑定到对应的死信交换机
@Bean("delayQueueA")
public Queue delayQueueA(){
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEA_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 6000);
return QueueBuilder.durable(DELAY_QUEUEA_NAME).withArguments(args).build();
}
// 声明延时队列B 延时 60s
// 并绑定到对应的死信交换机
@Bean("delayQueueB")
public Queue delayQueueB(){
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEB_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 60000);
return QueueBuilder.durable(DELAY_QUEUEB_NAME).withArguments(args).build();
}
// 声明死信队列A 用于接收延时10s处理的消息
@Bean("deadLetterQueueA")
public Queue deadLetterQueueA(){
return new Queue(DEAD_LETTER_QUEUEA_NAME);
}
// 声明死信队列B 用于接收延时60s处理的消息
@Bean("deadLetterQueueB")
public Queue deadLetterQueueB(){
return new Queue(DEAD_LETTER_QUEUEB_NAME);
}
// 声明延时队列A绑定关系
@Bean
public Binding delayBindingA(@Qualifier("delayQueueA") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEA_ROUTING_KEY);
}
// 声明业务队列B绑定关系
@Bean
public Binding delayBindingB(@Qualifier("delayQueueB") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEB_ROUTING_KEY);
}
// 声明死信队列A绑定关系
@Bean
public Binding deadLetterBindingA(@Qualifier("deadLetterQueueA") Queue queue,
@Qualifier("deadLetterExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEA_ROUTING_KEY);
}
// 声明死信队列B绑定关系
@Bean
public Binding deadLetterBindingB(@Qualifier("deadLetterQueueB") Queue queue,
@Qualifier("deadLetterExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEB_ROUTING_KEY);
}
}接下来，创建两个消费者，分别对两个死信队列的消息进行消费：
@Slf4j
@Component
public class DeadLetterQueueConsumer {
@RabbitListener(queues = DEAD_LETTER_QUEUEA_NAME)
public void receiveA(Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("当前时间：{},死信队列A收到消息：{}", new Date().toString(), msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
@RabbitListener(queues = DEAD_LETTER_QUEUEB_NAME)
public void receiveB(Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("当前时间：{},死信队列B收到消息：{}", new Date().toString(), msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}然后是消息的生产者：
@Component
public class DelayMessageSender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMsg(String msg, DelayTypeEnum type){
switch (type){
case DELAY_10s:
rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUEA_ROUTING_KEY, msg);
break;
case DELAY_60s:
rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUEB_ROUTING_KEY, msg);
break;
}
}
}接下来，我们暴露一个web接口来生产消息：
@Slf4j
@RequestMapping("rabbitmq")
@RestController
public class RabbitMQMsgController {
@Autowired
private DelayMessageSender sender;
@RequestMapping("sendmsg")
public void sendMsg(String msg, Integer delayType){
log.info("当前时间：{},收到请求，msg:{},delayType:{}", new Date(), msg, delayType);
sender.sendMsg(msg, Objects.requireNonNull(DelayTypeEnum.getDelayTypeEnumByValue(delayType)));
}
}准备就绪，启动！
打开rabbitMQ的管理后台，可以看到我们刚才创建的交换机和队列信息：
接下来，我们来发送几条消息，http://localhost:8080/rabbitmq/sendmsg?msg=testMsg1&delayType=1 http://localhost:8080/rabbitmq/sendmsg?msg=testMsg2&delayType=2
日志如下：
2019-07-28 16:02:19.813 INFO 3860 --- [nio-8080-exec-9] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 16:02:19 CST 2019,收到请求，msg:testMsg1,delayType:1
2019-07-28 16:02:19.815 INFO 3860 --- [nio-8080-exec-9] .l.DirectReplyToMessageListenerContainer : SimpleConsumer [queue=amq.rabbitmq.reply-to, consumerTag=amq.ctag-o-qPpkWIkRm73DIrOIVhig identity=766339] started
2019-07-28 16:02:25.829 INFO 3860 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 16:02:25 CST 2019,死信队列A收到消息：testMsg1
2019-07-28 16:02:41.326 INFO 3860 --- [nio-8080-exec-1] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 16:02:41 CST 2019,收到请求，msg:testMsg2,delayType:2
2019-07-28 16:03:41.329 INFO 3860 --- [ntContainer#0-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 16:03:41 CST 2019,死信队列B收到消息：testMsg2第一条消息在6s后变成了死信消息，然后被消费者消费掉，第二条消息在60s之后变成了死信消息，然后被消费掉，这样，一个还算ok的延时队列就打造完成了。
不过，等等，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列，这里只有6s和60s两个时间选项，如果需要一个小时后处理，那么就需要增加TTL为一个小时的队列，如果是预定会议室然后提前通知这样的场景，岂不是要增加无数个队列才能满足需求？？
嗯，仔细想想，事情并不简单。
显然，需要一种更通用的方案才能满足需求，那么就只能将TTL设置在消息属性里了。我们来试一试。
消息设置TTL
增加一个延时队列，用于接收设置为任意延时时长的消息，增加一个相应的死信队列和routingkey：
@Configuration
public class RabbitMQConfig {
public static final String DELAY_EXCHANGE_NAME = "delay.queue.demo.business.exchange";
public static final String DELAY_QUEUEC_NAME = "delay.queue.demo.business.queuec";
public static final String DELAY_QUEUEC_ROUTING_KEY = "delay.queue.demo.business.queuec.routingkey";
public static final String DEAD_LETTER_EXCHANGE = "delay.queue.demo.deadletter.exchange";
public static final String DEAD_LETTER_QUEUEC_ROUTING_KEY = "delay.queue.demo.deadletter.delay_anytime.routingkey";
public static final String DEAD_LETTER_QUEUEC_NAME = "delay.queue.demo.deadletter.queuec";
// 声明延时Exchange
@Bean("delayExchange")
public DirectExchange delayExchange(){
return new DirectExchange(DELAY_EXCHANGE_NAME);
}
// 声明死信Exchange
@Bean("deadLetterExchange")
public DirectExchange deadLetterExchange(){
return new DirectExchange(DEAD_LETTER_EXCHANGE);
}
// 声明延时队列C 不设置TTL
// 并绑定到对应的死信交换机
@Bean("delayQueueC")
public Queue delayQueueC(){
Map<String, Object> args = new HashMap<>(3);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEC_ROUTING_KEY);
return QueueBuilder.durable(DELAY_QUEUEC_NAME).withArguments(args).build();
}
// 声明死信队列C 用于接收延时任意时长处理的消息
@Bean("deadLetterQueueC")
public Queue deadLetterQueueC(){
return new Queue(DEAD_LETTER_QUEUEC_NAME);
}
// 声明延时列C绑定关系
@Bean
public Binding delayBindingC(@Qualifier("delayQueueC") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEC_ROUTING_KEY);
}
// 声明死信队列C绑定关系
@Bean
public Binding deadLetterBindingC(@Qualifier("deadLetterQueueC") Queue queue,
@Qualifier("deadLetterExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEC_ROUTING_KEY);
}
}增加一个死信队列C的消费者：
@RabbitListener(queues = DEAD_LETTER_QUEUEC_NAME)
public void receiveC(Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("当前时间：{},死信队列C收到消息：{}", new Date().toString(), msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}再次启动！然后访问：http://localhost:8080/rabbitmq/delayMsg?msg=testMsg1delayTime=5000 来生产消息，注意这里的单位是毫秒。
2019-07-28 16:45:07.033 INFO 31468 --- [nio-8080-exec-4] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 16:45:07 CST 2019,收到请求，msg:testMsg1,delayTime:5000
2019-07-28 16:45:11.694 INFO 31468 --- [nio-8080-exec-5] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 16:45:11 CST 2019,收到请求，msg:testMsg2,delayTime:5000
2019-07-28 16:45:12.048 INFO 31468 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 16:45:12 CST 2019,死信队列C收到消息：testMsg1
2019-07-28 16:45:16.709 INFO 31468 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 16:45:16 CST 2019,死信队列C收到消息：testMsg2看起来似乎没什么问题，但不要高兴的太早，在最开始的时候，就介绍过，如果使用在消息属性上设置TTL的方式，消息可能并不会按时“死亡“，因为RabbitMQ只会检查第一个消息是否过期，如果过期则丢到死信队列，索引如果第一个消息的延时时长很长，而第二个消息的延时时长很短，则第二个消息并不会优先得到执行。
实验一下：
2019-07-28 16:49:02.957 INFO 31468 --- [nio-8080-exec-8] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 16:49:02 CST 2019,收到请求，msg:longDelayedMsg,delayTime:20000
2019-07-28 16:49:10.671 INFO 31468 --- [nio-8080-exec-9] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 16:49:10 CST 2019,收到请求，msg:shortDelayedMsg,delayTime:2000
2019-07-28 16:49:22.969 INFO 31468 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 16:49:22 CST 2019,死信队列C收到消息：longDelayedMsg
2019-07-28 16:49:22.970 INFO 31468 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 16:49:22 CST 2019,死信队列C收到消息：shortDelayedMsg我们先发了一个延时时长为20s的消息，然后发了一个延时时长为2s的消息，结果显示，第二个消息会在等第一个消息成为死信后才会“死亡“。
上文中提到的问题，确实是一个硬伤，如果不能实现在消息粒度上添加TTL，并使其在设置的TTL时间及时死亡，就无法设计成一个通用的延时队列。
那如何解决这个问题呢？不要慌，安装一个插件即可：https://www.rabbitmq.com/community-plugins.html ，下载rabbitmq_delayed_message_exchange插件，然后解压放置到RabbitMQ的插件目录。
接下来，进入RabbitMQ的安装目录下的sbin目录，执行下面命令让该插件生效，然后重启RabbitMQ。
rabbitmq-plugins enable rabbitmq_delayed_message_exchange然后，我们再声明几个Bean：
@Configuration
public class DelayedRabbitMQConfig {
public static final String DELAYED_QUEUE_NAME = "delay.queue.demo.delay.queue";
public static final String DELAYED_EXCHANGE_NAME = "delay.queue.demo.delay.exchange";
public static final String DELAYED_ROUTING_KEY = "delay.queue.demo.delay.routingkey";
@Bean
public Queue immediateQueue() {
return new Queue(DELAYED_QUEUE_NAME);
}
@Bean
public CustomExchange customExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
}
@Bean
public Binding bindingNotify(@Qualifier("immediateQueue") Queue queue,
@Qualifier("customExchange") CustomExchange customExchange) {
return BindingBuilder.bind(queue).to(customExchange).with(DELAYED_ROUTING_KEY).noargs();
}
}controller层再添加一个入口：
@RequestMapping("delayMsg2")
public void delayMsg2(String msg, Integer delayTime) {
log.info("当前时间：{},收到请求，msg:{},delayTime:{}", new Date(), msg, delayTime);
sender.sendDelayMsg(msg, delayTime);
}消息生产者的代码也需要修改：
public void sendDelayMsg(String msg, Integer delayTime) {
rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, msg, a ->{
a.getMessageProperties().setDelay(delayTime);
return a;
});
}最后，再创建一个消费者：
@RabbitListener(queues = DELAYED_QUEUE_NAME)
public void receiveD(Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("当前时间：{},延时队列收到消息：{}", new Date().toString(), msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}一切准备就绪，启动！然后分别访问以下链接：
http://localhost:8080/rabbitmq/delayMsg2?msg=msg1&delayTime=20000
http://localhost:8080/rabbitmq/delayMsg2?msg=msg2&delayTime=2000日志如下：
2019-07-28 17:28:13.729 INFO 25804 --- [nio-8080-exec-2] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 17:28:13 CST 2019,收到请求，msg:msg1,delayTime:20000
2019-07-28 17:28:20.607 INFO 25804 --- [nio-8080-exec-1] c.m.d.controller.RabbitMQMsgController : 当前时间：Sun Jul 28 17:28:20 CST 2019,收到请求，msg:msg2,delayTime:2000
2019-07-28 17:28:22.624 INFO 25804 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 17:28:22 CST 2019,延时队列收到消息：msg2
2019-07-28 17:28:33.751 INFO 25804 --- [ntContainer#1-1] c.m.d.mq.DeadLetterQueueConsumer : 当前时间：Sun Jul 28 17:28:33 CST 2019,延时队列收到消息：msg1第二个消息被先消费掉了，符合预期。至此，RabbitMQ实现延时队列的部分就完结了。
延时队列在需要延时处理的场景下非常有用，使用RabbitMQ来实现延时队列可以很好的利用RabbitMQ的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如利用Java的DelayQueu，利用Redis的zset，利用Quartz或者利用kafka的时间轮，这些方式各有特点，但就像炉石传说一般，这些知识就好比手里的卡牌，知道的越多，可以用的卡牌也就越多，遇到问题便能游刃有余，所以需要大量的知识储备和经验积累才能打造出更出色的卡牌组合，让自己解决问题的能力得到更好的提升。
