RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式

一、什么是延时队列
延时队列顾名思义，即放置在该队列里面的消息是不需要立即消费的，而是等待一段时间之后取出消费。

二、延时队列应用于什么场景
场景一：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行一场处理。这是就可以使用延时队列将订单信息发送到延时队列。

场景二：用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作。这时候就可以将用户指令发送到延时队列，当指令设定的时间到了再将指令推送到智能设备。

Rabbitmq实现延时队列一般而言有两种形式：
第一种方式：利用两个特性： Time To Live(TTL)、Dead Letter Exchanges（DLX）
第二种方式：利用rabbitmq中的插件x-delay-message

三、第一种：利用TTL DLX实现延时队列的方式
AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。

1、Time To Live(TTL)

RabbitMQ可以针对Queue设置x-expires 或者针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)

A: 通过队列属性设置，队列中所有消息都有相同的过期时间。
B: 对消息进行单独设置，每条消息TTL可以不同。

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发送

用一个具体案例来实现第一种方式：用户下订单后，如何在一分钟没有支付就取消订单

package com.springboot.rabbitmq.example.demo5.config;

import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import lombok.extern.slf4j.Slf4j;

/**
* @method
* @author Mr yi
* @time 2019年6月23日
*/
@Configuration
@Slf4j
public class RabbitConfigDemo5 {

//队列名称
final static String queue = "queue_demo5";

//交换机名称
   final static String exchangeName = "deom5Exchange";

   // routingKey
   final static String routingKey = "keyDemo5";

   //死信消息队列名称
   final static String deal_queue = "deal_queue_demo5";

//死信交换机名称
   final static String deal_exchangeName = "deal_deom5Exchange";

   //死信 routingKey
   final static String dead_RoutingKey = "dead_routing_key";

   //死信队列交换机标识符
public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";

//死信队列交换机绑定键标识符
public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

@Autowired
private CachingConnectionFactory connectionFactory;

/**
*
* @method 定义队列(队列绑定一个死信交换机,并指定routing_key)
* @author Mr yi
* @time 2019年6月29日
* @return
*/
   @Bean
   public Queue queueDemo5() {
       // 将普通队列绑定到死信队列交换机上
Map<String, Object> args = new HashMap<>(2);
//args.put("x-message-ttl", 5 * 1000);//直接设置 Queue 延迟时间但如果直接给队列设置过期时间,这种做法不是很灵活
//这里采用发送消息动态设置延迟时间,这样我们可以灵活控制
args.put(DEAD_LETTER_QUEUE_KEY, deal_exchangeName);
args.put(DEAD_LETTER_ROUTING_KEY, dead_RoutingKey);
return new Queue(RabbitConfigDemo5.queue, true, false, false, args);
   }

//声明一个direct类型的交换机
   @Bean
   DirectExchange exchangeDemo5() {
       return new DirectExchange(RabbitConfigDemo5.exchangeName);
   }

//绑定Queue队列到交换机,并且指定routingKey
   @Bean
   Binding bindingDirectExchangeDemo5( ) {
       return BindingBuilder.bind(queueDemo5()).to(exchangeDemo5()).with(routingKey);
   }

   //创建配置死信队列
@Bean
public Queue deadQueue5() {
Queue queue = new Queue(deal_queue, true);
return queue;
}

//创建死信交换机
@Bean
public DirectExchange deadExchange5() {
return new DirectExchange(deal_exchangeName);
}

//死信队列与死信交换机绑定
@Bean
public Binding bindingDeadExchange5() {
return BindingBuilder.bind(deadQueue5()).to(deadExchange5()).with(dead_RoutingKey);
}

/** @Bean
public RabbitTemplate rabbitTemplate(){
       //若使用confirm-callback ，必须要配置publisherConfirms 为true
       connectionFactory.setPublisherConfirms(true);
       //若使用return-callback，必须要配置publisherReturns为true
connectionFactory.setPublisherReturns(true);
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true
// rabbitTemplate.setMandatory(true);

// 如果消息没有到exchange,则confirm回调,ack=false; 如果消息到达exchange,则confirm回调,ack=true
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
}else{
log.info("消息发送失败:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
}
}
});

//如果exchange到queue成功,则不回调return;如果exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
}
});
return rabbitTemplate;
}

**/

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
生产者产生订单后，将订单信息发送到rabbitmq 服务段，设置TTL 时间，如果超过了这个时间，还没有消费这个消息，那么就变为死信，发送到死信队列中。

这里利用死信的机制来巧妙的实现延时，我这里没有设置正常消费者，即生产者发送消息后，消息不会被消费，那么在指定时间后，变为死信，有与死信队列绑定的消费者来消费消息（判断订单是否已经成功支付）

package com.springboot.rabbitmq.example.demo5.producers;

import java.util.Date;
import java.util.UUID;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;
/**
*
* @method 生产者
* @author Mr yi
* @time 2019年6月19日
*/
@Component
@Slf4j
public class ProducersDemo5 {

@Autowired
private AmqpTemplate rabbitTemplate;

   /**
   * @method 生产者发送消息,direct模式下需要传递一个routingKey
   * @author Mr yi
   * @time 2019年6月19日
   * @throws Exception
   */
   public void send( ) throws Exception {

       log.info("【订单生成时间】" + new Date().toString() +"【1分钟后检查订单是否已经支付】" );

this.rabbitTemplate.convertAndSend("deom5Exchange", "keyDemo5", "订单实体类对象信息", message -> {
// 如果配置了 params.put("x-message-ttl", 5 * 1000); 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
message.getMessageProperties().setExpiration(1 * 1000 * 60 + "");
return message;
});

   }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
死信消息消费者

package com.springboot.rabbitmq.example.demo5.consumers;

import java.io.IOException;
import java.util.Date;
import java.util.Map;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

/**
*
* @method 死信消费者，消费从死信队列传来的消息
* @author Mr yi
* @time 2019年6月19日
*/
@Component
@Slf4j
public class ConsumersDemo5Deal {

   @RabbitListener(queues = "deal_queue_demo5")
public void process(String order, Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {

       log.info("【监听到延时队列消息】 - 【消费时间】 - [{}]- 【订单内容】 - [{}]", new Date(), order);
       // 判断订单是否已经支付，如果支付则；否则，取消订单（逻辑代码省略)

// 手动ack
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// 手动签收
channel.basicAck(deliveryTag, false);
System.out.println("执行结束....");

}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
测试

@Autowired
   private ProducersDemo5 producers;

   @RequestMapping("/send")
public String send() throws Exception {
   producers.send();
   return "success";
}
1
2
3
4
5
6
7
8
启动程序，测试

发现queue_demo5正常队列有一条消息处于待续状态

等待一分钟后，控制台输出

发现queue_demo5 消息已经被消费（发送到deal_queue_demo5死信队列了）

使用死信队列实现延时消息的缺点：

1）如果统一用队列来设置消息的TTL，当延时时间梯度比较多的话，比如1分钟，2分钟，5分钟，10分钟，20分钟，30分钟……需要创建很多交换机和队列来路由消息。
2）如果单独设置消息的TTL，则可能会造成队列中的消息阻塞——前一条消息没有出队（没有被消费），后面的消息无法投递。
3）可能存在一定的时间误差。

四、第二种：利用rabbitmq-delayed-message-exchange插件来实现延迟队列功能
插件下载地址：注意下载插件要和安装的rabbitmq版本一致，我这里下载的是3.7的
https://www.rabbitmq.com/community-plugins.html

下载解压后，得到一个.ez的压缩文件，找到rabbitmq安装目录的plugins文件夹，将解压的文件复制进去

重新启动rabbitmq ，输入命令
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

停止：net stop RabbitMQ
启动：net start RabbitMQ
1
2

配置类

package com.springboot.rabbitmq.example.demo6.config;

import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import lombok.extern.slf4j.Slf4j;

/**
* @method
* @author Mr yi
* @time 2019年6月23日
*/
@Configuration
@Slf4j
public class RabbitConfigDemo6 {

//队列名称
final static String queue = "queue_demo6";

//交换机名称
   final static String exchangeName = "deom6Exchange";

   // routingKey
   final static String routingKey = "keyDemo6";

@Autowired
private CachingConnectionFactory connectionFactory;

   @Bean
   public Queue queueDemo6() {
       // 第一个参数是创建的queue的名字，第二个参数是是否支持持久化
return new Queue(RabbitConfigDemo6.queue, true);
   }

@Bean
public CustomExchange delayExchange6() {
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-delayed-type", "direct");
return new CustomExchange(RabbitConfigDemo6.exchangeName, "x-delayed-message", true, false, args);
}

@Bean
public Binding bindingNotify6() {
return BindingBuilder.bind(queueDemo6()).to(delayExchange6()).with(RabbitConfigDemo6.routingKey).noargs();
}

**/

}

package com.springboot.rabbitmq.example.demo6.producers;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;
/**
*
* @method 生产者
* @author Mr yi
* @time 2019年6月19日
*/
@Component
@Slf4j
public class ProducersDemo6 {

@Autowired
private AmqpTemplate rabbitTemplate;

   /**
   * @method 生产者发送消息,direct模式下需要传递一个routingKey
   * @author Mr yi
   * @time 2019年6月19日
   * @throws Exception
   */
   public void send( ) throws Exception {

       log.info("【订单生成时间】" + new Date().toString() +"【1分钟后检查订单是否已经支付】" );

       this.rabbitTemplate.convertAndSend("deom6Exchange", "keyDemo6", "订单实体类对象信息", new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setDelay(1 * 1000 * 60 );
return message;
}
});


   }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
消费者，消费者一分钟后得到生产者发送的消息

package com.springboot.rabbitmq.example.demo6.consumers;

import java.io.IOException;
import java.util.Date;
import java.util.Map;

import lombok.extern.slf4j.Slf4j;

/**
*
* @method 消费者
* @author Mr yi
* @time 2019年6月19日
*/
@Component
@Slf4j
public class ConsumersDemo6 {

   @RabbitListener(queues = "queue_demo6")
public void process(String order, Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {

       log.info("【监听到延时队列消息】 - 【消费时间】 - [{}]- 【订单内容】 - [{}]", new Date(), order);
       // 判断订单是否已经支付，如果支付则；否则，取消订单（逻辑代码省略)

// 手动ack
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// 手动签收
channel.basicAck(deliveryTag, false);
System.out.println("执行结束....");

}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
测试

@Autowired
   private ProducersDemo6 producers;

   @RequestMapping("/send")
public String send() throws Exception {
   producers.send();
   return "success";
}
1
2
3
4
5
6
7
8
启动程序，执行方法

控制台输出

rabbitmq服务端，queue_demo6 其中并没有消息进入就绪状态，这一点也是和第一种方式（使用死信）的区别优势所在。

等待一分钟后，消费者接受到消息控制台

源码下载：https://download.csdn.net/download/qq_29914837/11264460

如果你觉得本篇文章对你有所帮助的话，麻烦请点击头像右边的关注按钮，谢谢！

技术在交流中进步，知识在分享中传播
————————————————
版权声明：本文为CSDN博主「互联网叫兽」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/qq_29914837/article/details/94070677

RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式相关推荐

RabbitMQ（六）——Spring boot中消费消息的两种方式
前言上一篇博客中,我们只是简单总结了Spring boot中整合RabbitMQ的操作,针对消息消费的两种方式只是简单给了一个实例,这篇博客,我们进一步总结关于Spring boot消息消费的相关功 ...
延时队列的几种实现方式
延时队列的几种实现方式何为延迟队列? 顾名思义,首先它要具有队列的特性,再给它附加一个延迟消费队列消息的功能,也就是说可以指定队列中的消息在哪个时间点被消费. 延时队列能做什么? 延时队列多用于需要 ...
iOS:延时执行的三种方式
延时执行的三种方式:performSelectorXXX方法.GCD中延时函数.创建定时器第一种方式:NSObject分类当中的方法,延迟一段时间调用某一个方法 @interface NSObjec ...
python中延时函数_python开发--Python实现延时操作的几种方式
1. time.sleep 2. sched.scheduler 3. threading.Timer 4. 借助其他程序 celery redis延时队列在日常的开发中,往往会遇到这样的需求,需要 ...
分享笔记RabbitMQ高级之消息限流与延时队列
楔子本篇是消息队列RabbitMQ的第五弹. 上篇本来打算讲述RabbitMQ的一些高级用法: 如何保证消息的可靠性? 消息队列如何进行限流? 如何设置延时队列进行延时消费? 最终因为篇幅缘故,上篇 ...
RabbitMQ延迟消息场景分析以及实现两种方式（SpringBoot）
使用场景,不限于下面用户下订单结束,如果用户未及时支付,后续需要取消订单,可以怎么做?定时任务可以做,但是不能接近实时或者消耗数据库性能太大. [数据库定时任务方案]:定时任务可以做到订单的状态的改 ...
Android 延时操作的三种方式
在Android开发中我们可能会有延时执行某个操作的需求,例如我们启动应用的时候,一开始呈现的是引导页面,3秒后进入主界面,这就是一个延时操作. 下面是实现延时操作的三种方法: 一.使用线程的休眠实现 ...
RabbitMQ几种工作模式，实现延时消息的两种案例
点击上方蓝色字体,选择"标星公众号" 优质文章,第一时间送达 1.死信队列 1.1消息什么时候变为死信(dead-letter) 消息被否定接收,消费者使用basic.reject ...
RabbitMQ学习(七)_RabbitMQ Consumer获取消息的两种方式(poll,subscribe)解析
RabbitMQ中consumer通过建立到queue的连接,创建channel对象,通过channel通道获取message, Consumer可以声明式的以API轮询poll的方式主动从queue ...

RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式

RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式相关推荐

最新文章

热门文章