RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式
一、什么是延時隊列
延時隊列顧名思義，即放置在該隊列里面的消息是不需要立即消費的，而是等待一段時間之后取出消費。
二、延時隊列應(yīng)用于什么場景
場景一：在訂單系統(tǒng)中，一個用戶下單之后通常有30分鐘的時間進行支付，如果30分鐘之內(nèi)沒有支付成功，那么這個訂單將進行一場處理。這是就可以使用延時隊列將訂單信息發(fā)送到延時隊列。
場景二：用戶希望通過手機遠程遙控家里的智能設(shè)備在指定的時間進行工作。這時候就可以將用戶指令發(fā)送到延時隊列，當(dāng)指令設(shè)定的時間到了再將指令推送到智能設(shè)備。
Rabbitmq實現(xiàn)延時隊列一般而言有兩種形式：
第一種方式：利用兩個特性： Time To Live(TTL)、Dead Letter Exchanges（DLX）
第二種方式：利用rabbitmq中的插件x-delay-message
三、第一種：利用TTL DLX實現(xiàn)延時隊列的方式
AMQP協(xié)議和RabbitMQ隊列本身沒有直接支持延遲隊列功能，但是可以通過以下特性模擬出延遲隊列的功能。
1、Time To Live(TTL)
RabbitMQ可以針對Queue設(shè)置x-expires 或者 針對Message設(shè)置 x-message-ttl，來控制消息的生存時間，如果超時(兩者同時設(shè)置以最先到期的時間為準(zhǔn))，則消息變?yōu)閐ead letter(死信)
A: 通過隊列屬性設(shè)置，隊列中所有消息都有相同的過期時間。
B: 對消息進行單獨設(shè)置，每條消息TTL可以不同。
2、Dead Letter Exchanges（DLX）
RabbitMQ的Queue可以配置x-dead-letter-exchange和x-dead-letter-routing-key（可選）兩個參數(shù)，如果隊列內(nèi)出現(xiàn)了dead letter，則按照這兩個參數(shù)重新路由轉(zhuǎn)發(fā)到指定的隊列。
x-dead-letter-exchange：出現(xiàn)dead letter之后將dead letter重新發(fā)送到指定exchange
x-dead-letter-routing-key：出現(xiàn)dead letter之后將dead letter重新按照指定的routing-key發(fā)送
用一個具體案例來實現(xiàn)第一種方式：用戶下訂單后，如何在一分鐘沒有支付就取消訂單
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";
?? ?
?? ?//死信隊列 交換機標(biāo)識符
? ? public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
? ??
? ? //死信隊列交換機綁定鍵標(biāo)識符
? ? 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);//直接設(shè)置 Queue 延遲時間 但如果直接給隊列設(shè)置過期時間,這種做法不是很靈活
? ? ? ? //這里采用發(fā)送消息動態(tài)設(shè)置延遲時間,這樣我們可以靈活控制
? ? ? ? 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);
?? ?}
?? ?
?? ?//創(chuàng)建配置死信隊列
? ? @Bean
? ? public Queue deadQueue5() {
? ? ? ? Queue queue = new Queue(deal_queue, true);
? ? ? ? return queue;
? ? }
?? ?
? ? //創(chuàng)建死信交換機
? ? ?@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時必須設(shè)置mandatory為true，或者在配置中設(shè)置mandatory-expression的值為true
? ? ? ? ?// rabbitTemplate.setMandatory(true);
? ?
? ? ? ? ? // 如果消息沒有到exchange,則confirm回調(diào),ack=false; 如果消息到達exchange,則confirm回調(diào),ack=true
? ? ? ? ? rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? public void confirm(CorrelationData correlationData, boolean ack, String cause) {
? ? ? ? ? ? ? ? ? if(ack){
? ? ? ? ? ? ? ? ? ? ? log.info("消息發(fā)送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
? ? ? ? ? ? ? ? ? }else{
? ? ? ? ? ? ? ? ? ? ? log.info("消息發(fā)送失敗:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? }
? ? ? ? ? });
? ? ? ? ??
? ? ? ? ? //如果exchange到queue成功,則不回調(diào)return;如果exchange到queue失敗,則回調(diào)return(需設(shè)置mandatory=true,否則不回回調(diào),消息就丟了)
? ? ? ? ? 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
生產(chǎn)者產(chǎn)生訂單后，將訂單信息發(fā)送到rabbitmq 服務(wù)段，設(shè)置TTL 時間，如果超過了這個時間，還沒有消費這個消息，那么就變?yōu)樗佬?#xff0c;發(fā)送到死信隊列中。
這里利用死信的機制來巧妙的實現(xiàn)延時，我這里沒有設(shè)置正常消費者，即生產(chǎn)者發(fā)送消息后，消息不會被消費，那么在指定時間后，變?yōu)樗佬?#xff0c;有與死信隊列綁定的消費者來消費消息（判斷訂單是否已經(jīng)成功支付）
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 生產(chǎn)者
?* @author Mr yi
?* @time 2019年6月19日
?*/
@Component
@Slf4j
public class ProducersDemo5 ?{
?
? ? @Autowired
? ? private AmqpTemplate rabbitTemplate;
? ??
?? ?/**
?? ? * @method 生產(chǎn)者發(fā)送消息,direct模式下需要傳遞一個routingKey
?? ? * @author Mr yi
?? ? * @time 2019年6月19日
?? ? * @throws Exception
?? ? */
?? ?public void send( ) throws Exception {
?? ??? ?
?? ??? ?log.info("【訂單生成時間】" + new Date().toString() +"【1分鐘后檢查訂單是否已經(jīng)支付】" ?);
? ? ? ??
? ? ? ? this.rabbitTemplate.convertAndSend("deom5Exchange", "keyDemo5", "訂單實體類對象信息", message -> {
? ? ? ? ? ? // 如果配置了 params.put("x-message-ttl", 5 * 1000); 那么這一句也可以省略,具體根據(jù)業(yè)務(wù)需要是聲明 Queue 的時候就指定好延遲時間還是在發(fā)送自己控制時間
? ? ? ? ? ? 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("【 監(jiān)聽到延時隊列消息】 - 【消費時間】 - [{}]- 【訂單內(nèi)容】 - [{}]", ?new Date(), order);?
?? ??? ?// 判斷訂單是否已經(jīng)支付，如果支付則；否則，取消訂單（邏輯代碼省略)
?? ??? ?
? ? ? ? // 手動ack
? ? ? ? Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
? ? ? ? // 手動簽收
? ? ? ? channel.basicAck(deliveryTag, false);
? ? ? ? System.out.println("執(zhí)行結(jié)束....");
? ? ? ??
? ? }
?? ?
}
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
啟動程序，測試
發(fā)現(xiàn)queue_demo5正常隊列有一條消息處于待續(xù)狀態(tài)
等待一分鐘后，控制臺輸出
發(fā)現(xiàn)queue_demo5 消息已經(jīng)被消費（發(fā)送到deal_queue_demo5死信隊列了）
使用死信隊列實現(xiàn)延時消息的缺點：
1） 如果統(tǒng)一用隊列來設(shè)置消息的TTL，當(dāng)延時時間梯度比較多的話，比如1分鐘，2分鐘，5分鐘，10分鐘，20分鐘，30分鐘……需要創(chuàng)建很多交換機和隊列來路由消息。
2） 如果單獨設(shè)置消息的TTL，則可能會造成隊列中的消息阻塞——前一條消息沒有出隊（沒有被消費），后面的消息無法投遞。
3） 可能存在一定的時間誤差。
四、第二種：利用rabbitmq-delayed-message-exchange插件來實現(xiàn)延遲隊列功能
插件下載地址：注意下載插件要和安裝的rabbitmq版本一致，我這里下載的是3.7的
https://www.rabbitmq.com/community-plugins.html
下載解壓后，得到一個.ez的壓縮文件，找到rabbitmq安裝目錄的plugins文件夾，將解壓的文件復(fù)制進去
重新啟動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() {
?? ??? ?// 第一個參數(shù)是創(chuàng)建的queue的名字，第二個參數(shù)是是否支持持久化
? ? ? ? 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();
? ? }
?
?? ??
/** ? ? ?@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時必須設(shè)置mandatory為true，或者在配置中設(shè)置mandatory-expression的值為true
? ? ? ? ?// rabbitTemplate.setMandatory(true);
? ?
? ? ? ? ? // 如果消息沒有到exchange,則confirm回調(diào),ack=false; 如果消息到達exchange,則confirm回調(diào),ack=true
? ? ? ? ? rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? public void confirm(CorrelationData correlationData, boolean ack, String cause) {
? ? ? ? ? ? ? ? ? if(ack){
? ? ? ? ? ? ? ? ? ? ? log.info("消息發(fā)送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
? ? ? ? ? ? ? ? ? }else{
? ? ? ? ? ? ? ? ? ? ? log.info("消息發(fā)送失敗:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? }
? ? ? ? ? });
? ? ? ? ??
? ? ? ? ? //如果exchange到queue成功,則不回調(diào)return;如果exchange到queue失敗,則回調(diào)return(需設(shè)置mandatory=true,否則不回回調(diào),消息就丟了)
? ? ? ? ? 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
生產(chǎn)者，設(shè)置setDelay(1 * 1000 * 60 ); 延時 1分鐘
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 生產(chǎn)者
?* @author Mr yi
?* @time 2019年6月19日
?*/
@Component
@Slf4j
public class ProducersDemo6 {
?
? ? @Autowired
? ? private AmqpTemplate rabbitTemplate;
? ??
?? ?/**
?? ? * @method 生產(chǎn)者發(fā)送消息,direct模式下需要傳遞一個routingKey
?? ? * @author Mr yi
?? ? * @time 2019年6月19日
?? ? * @throws Exception
?? ? */
?? ?public void send( ) throws Exception {
?? ??? ?
?? ??? ?log.info("【訂單生成時間】" + new Date().toString() +"【1分鐘后檢查訂單是否已經(jīng)支付】" ?);
?? ??? ?
?? ??? ?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
消費者，消費者一分鐘后得到生產(chǎn)者發(fā)送的消息
package com.springboot.rabbitmq.example.demo6.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 ConsumersDemo6 {
?? ?
?? ?@RabbitListener(queues = "queue_demo6")
? ? public void process(String order, ?Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
?? ?
?? ??? ?log.info("【 監(jiān)聽到延時隊列消息】 - 【消費時間】 - [{}]- 【訂單內(nèi)容】 - [{}]", ?new Date(), order);?
?? ??? ?// 判斷訂單是否已經(jīng)支付，如果支付則；否則，取消訂單（邏輯代碼省略)
?? ??? ?
? ? ? ? // 手動ack
? ? ? ? Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
? ? ? ? // 手動簽收
? ? ? ? channel.basicAck(deliveryTag, false);
? ? ? ? System.out.println("執(zhí)行結(jié)束....");
? ? ? ??
? ? }
?? ?
}
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
啟動程序，執(zhí)行方法
控制臺輸出
rabbitmq服務(wù)端，queue_demo6 其中并沒有消息進入就緒狀態(tài)，這一點也是和第一種方式（使用死信）的區(qū)別優(yōu)勢所在。
等待一分鐘后，消費者接受到消息控制臺
源碼下載：https://download.csdn.net/download/qq_29914837/11264460
如果你覺得本篇文章對你有所幫助的話，麻煩請點擊頭像右邊的關(guān)注按鈕，謝謝！
技術(shù)在交流中進步，知識在分享中傳播
————————————————
版權(quán)聲明：本文為CSDN博主「互聯(lián)網(wǎng)叫獸」的原創(chuàng)文章，遵循CC 4.0 BY-SA版權(quán)協(xié)議，轉(zhuǎn)載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/qq_29914837/article/details/94070677
總結(jié)
以上是生活随笔為你收集整理的RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 【RabbitMQ】一文带你搞定Rabb
- 下一篇： cron表达式详解 Elastic-J