手把手帶你掌握SpringBoot RabbitMQ延遲隊(duì)列
1. 簡(jiǎn)介
我們?cè)谏弦黄┪闹羞z留了一個(gè)小問題,就是雖然TTL + DLX
能實(shí)現(xiàn)延遲隊(duì)列的功能,但是有兩個(gè)問題。
首先業(yè)務(wù)場(chǎng)景為:比如海底撈預(yù)約,每個(gè)人預(yù)約的時(shí)間段不一致,有個(gè)可能一個(gè)小時(shí)后,有的可能三個(gè)小時(shí)等,當(dāng)快到預(yù)約時(shí)間點(diǎn)需要給用戶進(jìn)行短信通知。
通過給Queue
設(shè)置過期時(shí)間的方式不現(xiàn)實(shí),因?yàn)楹苡锌赡苊織l記錄的過期時(shí)間都不一樣,不可能設(shè)置那么多的Queue
。直接給Message
設(shè)置過期時(shí)間,這種方式也不好,因?yàn)檫@種方式是當(dāng)該消息在隊(duì)列頭部時(shí)(消費(fèi)時(shí)),才會(huì)單獨(dú)判斷這一消息是否過期。例:現(xiàn)在有兩條消息,第一條消息過期時(shí)間為30s,而第二條消息過期時(shí)間為15s,當(dāng)過了15秒后,第二條消息不會(huì)立即過期,而是要等第一條消息被消費(fèi)后,第二條消息被消費(fèi)時(shí),才會(huì)判斷是否過期,也就是等到第二條消息投往DLX
已經(jīng)過去45s了。
這也就拋出了本章主題:延遲隊(duì)列
。
RabbitMQ默認(rèn)沒有提供延遲隊(duì)列
功能,而是要通過插件提供的x-delayed-message
(延遲交換機(jī))來(lái)實(shí)現(xiàn)。
延遲隊(duì)列
:用戶可以使用該類型聲明一個(gè)交換,x-delayed-message
然后使用自定義標(biāo)頭發(fā)布消息,x-delay
以毫秒為單位表示消息的延遲時(shí)間。消息將在x-delay
毫秒后傳遞到相應(yīng)的隊(duì)列。
2. 安裝插件
官方插件地址:https://www.rabbitmq.com/community-plugins.html
找到插件rabbitmq_delayed_message_exchange
,進(jìn)入GitHub下載本地RabbitMQ對(duì)應(yīng)的插件版本(下載.ez
文件)。
我這里下載的是3.8.9版本,如圖:
下載到本地后將文件放置RabbitMQ的plugins目錄。
我這里本地是使用docker-compose安裝的服務(wù),image
為rabbitmq:3.8.3-management
(雖然版本沒對(duì)起來(lái),但是測(cè)試能用,但是使用3.9的版本會(huì)報(bào)錯(cuò),插件安裝失?。┌惭b的服務(wù),操作步驟如下:
1.將下載好的文件放置RabbitMQ插件目錄
rabbitmq:容器服務(wù)名
$ docker cp /Users/ludangxin/Downloads/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq:/opt/rabbitmq/plugins/
2.進(jìn)入容器
$ docker exec -it rabbitmq /bin/bash
3.查看現(xiàn)有的插件列表
$ rabbitmq-plugins list # 輸出部分內(nèi)容如下 [E*] = 明確啟用; e = 隱式啟用 [ ] rabbitmq_amqp1_0 3.8.3 [ ] rabbitmq_auth_backend_cache 3.8.3 [ ] rabbitmq_auth_backend_http 3.8.3 [ ] rabbitmq_auth_backend_ldap 3.8.3 [ ] rabbitmq_auth_backend_oauth2 3.8.3 [ ] rabbitmq_auth_mechanism_ssl 3.8.3 [ ] rabbitmq_consistent_hash_exchange 3.8.3 [ ] rabbitmq_event_exchange 3.8.3 [ ] rabbitmq_federation 3.8.3 [ ] rabbitmq_federation_management 3.8.3 [ ] rabbitmq_jms_topic_exchange 3.8.3 [E*] rabbitmq_management 3.8.3 [e*] rabbitmq_management_agent 3.8.3 [ ] rabbitmq_mqtt 3.8.3
4.啟用插件
$ rabbitmq-plugins enable rabbitmq_delayed_message_exchange
再次查看安裝列表就有了rabbitmq_delayed_message_exchange
安裝完畢后登陸RabbitMQ控制臺(tái)查看,會(huì)發(fā)現(xiàn)多了個(gè)x-delayed-message
類型的Exchange。
3. 實(shí)現(xiàn)延遲隊(duì)列
3.1 引入所需依賴
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency>
3.2 application.yaml
spring: rabbitmq: host: localhost port: 5672 # rabbit 默認(rèn)的虛擬主機(jī) virtual-host: / # rabbit 用戶名密碼 username: admin password: admin123
3.3 RabbitConfig
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.CustomExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.HashMap; import java.util.Map; /** * 延遲隊(duì)列配置 * * @author ludangxin * @date 2021/9/16 */ @Configuration public class RabbitDelayedConfig { public static final String QUEUE_NAME_DELAYED = "DELAY.QUEUE"; public static final String EXCHANGE_NAME_DELAYED = "DELAY.EXCHANGE"; public static final String ROUTING_KEY_DELAYED = "DELAY.#"; @Bean(QUEUE_NAME_DELAYED) public Queue queue() { return QueueBuilder.durable(QUEUE_NAME_DELAYED).build(); } @Bean(EXCHANGE_NAME_DELAYED) public CustomExchange exchange() { Map<String, Object> arguments = new HashMap<>(1); // 在這里聲明一個(gè)主題類型的延遲隊(duì)列,當(dāng)然其他類型的也可以。 arguments.put("x-delayed-type", "topic"); return new CustomExchange(EXCHANGE_NAME_DELAYED, "x-delayed-message", true, false, arguments); } @Bean public Binding bindingNotify(@Qualifier(QUEUE_NAME_DELAYED) Queue queue, @Qualifier(EXCHANGE_NAME_DELAYED) CustomExchange customExchange) { return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_DELAYED).noargs(); } }
3.4 Producer
import com.ldx.rabbitmq.config.RabbitDelayedConfig; import org.springframework.amqp.core.Message; 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 java.time.LocalDateTime; import java.time.format.DateTimeFormatter; /** * 延遲消息生產(chǎn)者 * * @author ludangxin * @date 2021/9/9 */ @Component public class DelayProducer { @Autowired private RabbitTemplate rabbitTemplate; public void sendDelayedMsg(String msg, Integer delay) { MessageProperties mp = new MessageProperties(); // 設(shè)置過期時(shí)間 mp.setDelay(delay); Message message = new Message(msg.getBytes(), mp); rabbitTemplate.convertAndSend(RabbitDelayedConfig.EXCHANGE_NAME_DELAYED, "DELAY.MSG", message); } }
3.5 Consumer
import com.ldx.rabbitmq.config.RabbitDelayedConfig; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; /** * 延遲消息消費(fèi)者 * * @author ludangxin * @date 2021/9/9 */ @Slf4j @Component public class DelayConsumer { @RabbitListener(queues = {RabbitDelayedConfig.QUEUE_NAME_DELAYED}) public void delayQueue(Message message){ log.info(new String(message.getBody()) + ",結(jié)束時(shí)間為:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); } }
3.6 測(cè)試代碼
@Autowired private DelayProducer delayProducer; @Test @SneakyThrows public void sendDelayedMsg() { for(int i = 16; i >= 10; i --) { String msg = "我將在" + i + "s后過期,開始時(shí)間為:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); delayProducer.sendDelayedMsg(msg,i * 1000); } // 使進(jìn)程阻塞,方便Consumer監(jiān)聽輸出Message System.in.read(); }
3.7 啟動(dòng)測(cè)試
啟動(dòng)測(cè)試代碼,輸出內(nèi)容如下:
從日志內(nèi)容可以看出,消息存活了30s,符合預(yù)期。
2021-09-16 23:40:10.806 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在10s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:10
2021-09-16 23:40:11.792 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在11s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:11
2021-09-16 23:40:12.791 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在12s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:12
2021-09-16 23:40:13.791 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在13s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:13
2021-09-16 23:40:14.788 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在14s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:14
2021-09-16 23:40:15.785 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在15s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:15
2021-09-16 23:40:16.785 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我將在16s后過期,開始時(shí)間為:2021-09-16 23:40:00,結(jié)束時(shí)間為:2021-09-16 23:40:16
到此這篇關(guān)于手把手帶你掌握SpringBoot RabbitMQ延遲隊(duì)列的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMQ 延遲隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
java JVM原理與常識(shí)知識(shí)點(diǎn)
在本文中小編給大家分享的是關(guān)于java的JVM原理和java常識(shí),有興趣的朋友們可以學(xué)習(xí)下2018-12-12springboot解決Class path contains multiple 
這篇文章主要介紹了springboot解決Class path contains multiple SLF4J bindings問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-07-07java實(shí)現(xiàn)文件編碼轉(zhuǎn)換的方法
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)文件編碼轉(zhuǎn)換的方法,分享一個(gè)文件編碼轉(zhuǎn)換的工具類,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2018-05-05IDEA報(bào)java:?java.lang.OutOfMemoryError:?Java?heap?space錯(cuò)誤
這篇文章主要給大家介紹了關(guān)于IDEA報(bào)java:?java.lang.OutOfMemoryError:?Java?heap?space錯(cuò)誤的解決辦法,文中將解決的辦法介紹的非常詳細(xì),需要的朋友可以參考下2024-01-01nacos配置中心遠(yuǎn)程調(diào)用讀取不到配置文件的解決
這篇文章主要介紹了nacos配置中心遠(yuǎn)程調(diào)用讀取不到配置文件的解決方案,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教。2022-01-01