SpringBoot+RabbitMQ 死信隊(duì)列

程序員的成長(zhǎng)之路

互聯(lián)網(wǎng)/程序員/技術(shù)/資料共享?

閱讀本文大概需要 4.5 分鐘。

來(lái)自：https://blog.csdn.net/m0_46144826

前言

死信：無(wú)法被消費(fèi)的消息，稱為死信。

如果死信一直留在隊(duì)列中，會(huì)導(dǎo)致一直被消費(fèi)，卻從不消費(fèi)成功。

所以我們專門開辟了一個(gè)來(lái)存放死信的隊(duì)列，叫死信隊(duì)列（DLX，dead-letter-exchange）。

死信的幾種來(lái)源：

• 消息 TTL 過(guò)期（time to live，存活時(shí)間，可以用在限時(shí)支付消息）
• 隊(duì)列達(dá)到最大長(zhǎng)度（隊(duì)列滿了，無(wú)法路由到該隊(duì)列）
• 消息被拒絕（?basic.reject / basic.nack?），并且?requeue = false

環(huán)境準(zhǔn)備配置

準(zhǔn)備 MQ 的隊(duì)列和環(huán)境：

• 正常交換機(jī)
• 正常隊(duì)列（最長(zhǎng)隊(duì)列 5） ---- 正常消費(fèi)者，拒絕消息
• ttl 隊(duì)列（過(guò)期時(shí)間 60 秒） ---- 沒有消費(fèi)者
• 死信交換機(jī)
• 死信隊(duì)列

主要配置文件如下：

@Configuration
public?class?DeadConfig?{

????/*?正常配置?**********************************************************************************************************/

????/**
?????*?正常交換機(jī)，開啟持久化
?????*/
????@Bean
????DirectExchange?normalExchange()?{
????????return?new?DirectExchange("normalExchange",?true,?false);
????}

????@Bean
????public?Queue?normalQueue()?{
????????// durable:?是否持久化,默認(rèn)是false,持久化隊(duì)列：會(huì)被存儲(chǔ)在磁盤上，當(dāng)消息代理重啟時(shí)仍然存在，暫存隊(duì)列：當(dāng)前連接有效
????????// exclusive:?默認(rèn)也是false，只能被當(dāng)前創(chuàng)建的連接使用，而且當(dāng)連接關(guān)閉后隊(duì)列即被刪除。此參考優(yōu)先級(jí)高于durable
????????// autoDelete:?是否自動(dòng)刪除，當(dāng)沒有生產(chǎn)者或者消費(fèi)者使用此隊(duì)列，該隊(duì)列會(huì)自動(dòng)刪除。
????????Map?args?=?deadQueueArgs();
????????//?隊(duì)列設(shè)置最大長(zhǎng)度
????????args.put("x-max-length",?5);
????????return?new?Queue("normalQueue",?true,?false,?false,?args);
????}

????@Bean
????public?Queue?ttlQueue()?{
????????Map?args?=?deadQueueArgs();
????????//?隊(duì)列設(shè)置消息過(guò)期時(shí)間?60?秒
????????args.put("x-message-ttl",?60?*?1000);
????????return?new?Queue("ttlQueue",?true,?false,?false,?args);
????}

????@Bean
????Binding?normalRouteBinding()?{
????????return?BindingBuilder.bind(normalQueue()).to(normalExchange()).with("normalRouting");
????}

????@Bean
????Binding?ttlRouteBinding()?{
????????return?BindingBuilder.bind(ttlQueue()).to(normalExchange()).with("ttlRouting");
????}

????/*?死信配置?**********************************************************************************************************/

????/**
?????*?死信交換機(jī)
?????*/
????@Bean
????DirectExchange?deadExchange()?{
????????return?new?DirectExchange("deadExchange",?true,?false);
????}

????/**
?????*?死信隊(duì)列
?????*/
????@Bean
????public?Queue?deadQueue()?{
????????return?new?Queue("deadQueue",?true,?false,?false);
????}

????@Bean
????Binding?deadRouteBinding()?{
????????return?BindingBuilder.bind(deadQueue()).to(deadExchange()).with("deadRouting");
????}

????/**
?????*?轉(zhuǎn)發(fā)到?死信隊(duì)列，配置參數(shù)
?????*/
????private?Map?deadQueueArgs()?{
????????Map?map?=?new?HashMap<>();
????????//?綁定該隊(duì)列到私信交換機(jī)
????????map.put("x-dead-letter-exchange",?"deadExchange");
????????map.put("x-dead-letter-routing-key",?"deadRouting");
????????return?map;
????}

}

arguments 具體參數(shù)如下：

隊(duì)列達(dá)到最大長(zhǎng)度

首先測(cè)試最簡(jiǎn)單的，沒有消費(fèi)者。

調(diào)用6次正常隊(duì)列的生產(chǎn)方法。

?/**
??*?正常消息隊(duì)列，隊(duì)列最大長(zhǎng)度5
??*/
?@GetMapping("/normalQueue")
?public?String?normalQueue()?{

?????Map?map?=?new?HashMap<>();
?????map.put("messageId",?String.valueOf(UUID.randomUUID()));
?????map.put("data",?System.currentTimeMillis()?+?",?正常隊(duì)列消息，最大長(zhǎng)度?5");

?????rabbitTemplate.convertAndSend("normalExchange",?"normalRouting",?map,?new?CorrelationData());
?????return?JSONObject.toJSONString(map);
?}

MQ 結(jié)果如下：

消息 TTL 過(guò)期

消息的TTL 指的是消息的存活時(shí)間，我們可以通過(guò)設(shè)置消息的TTL或者隊(duì)列的TTL來(lái)實(shí)現(xiàn)。

• 消息的TTL ：對(duì)于設(shè)置了過(guò)期時(shí)間屬性(expiration)的消息，消息如果在過(guò)期時(shí)間內(nèi)沒被消費(fèi)，會(huì)過(guò)期
• 隊(duì)列的TTL ：對(duì)于設(shè)置了過(guò)期時(shí)間屬性(x-message-ttl)的隊(duì)列，所有路由到這個(gè)隊(duì)列的消息，都會(huì)設(shè)置上這個(gè)過(guò)期時(shí)間

兩種配置都行，一般都用在定時(shí)任務(wù)，限時(shí)支付這種地方。

?/**
??*?消息?TTL,?time?to?live
??*/
?@GetMapping("/ttlToDead")
?public?String?ttlToDead()?{

?????Map?map?=?new?HashMap<>();
?????map.put("messageId",?String.valueOf(UUID.randomUUID()));
?????map.put("data",?System.currentTimeMillis()?+?",?ttl隊(duì)列消息");

?????rabbitTemplate.convertAndSend("normalExchange",?"ttlRouting",?map,?new?CorrelationData());
?????return?JSONObject.toJSONString(map);
?}

發(fā)送后：

等待過(guò)期后：

Demo 中只是為了方便，代碼中盡量使用 消息TTL，不要用 隊(duì)列TTL

拒絕消息

正常隊(duì)列消費(fèi)后拒絕消息，并且不進(jìn)行重新入隊(duì)：

@Component
@RabbitListener(queues?=?"normalQueue")
public?class?NormalConsumer?{
????@RabbitHandler
????public?void?process(Map?message,?Channel?channel,?Message?mqMsg)?throws?IOException?{
????????System.out.println("收到消息，并拒絕重新入隊(duì)?:?"?+?message.toString());
????????channel.basicReject(mqMsg.getMessageProperties().getDeliveryTag(),?false);
????}
}

MQ 控制臺(tái)：

死信隊(duì)列消費(fèi)：

@Component
@RabbitListener(queues?=?"deadQueue")
public?class?DeadConsumer?{
????@RabbitHandler
????public?void?process(Map?message,?Channel?channel,?Message?mqMsg)?throws?IOException?{
????????System.out.println("死信隊(duì)列收到消息?:?"?+?message.toString());
????????channel.basicAck(mqMsg.getMessageProperties().getDeliveryTag(),?false);
????}
}

消息順序和實(shí)驗(yàn)一致：

死信隊(duì)列收到消息?:?{data=1631534291765,?正常隊(duì)列消息，最大長(zhǎng)度?5,?messageId=bce3888b-da38-4299-ac88-d22cbe164739}
死信隊(duì)列收到消息?:?{data=1631535222745,?ttl隊(duì)列消息,?messageId=a4617445-5aab-4fac-aec7-5709ea699598}
死信隊(duì)列收到消息?:?{data=1631534503765,?正常隊(duì)列消息，最大長(zhǎng)度?5,?messageId=b65ecaab-5ce7-4597-a32c-c90b67ec46da}
死信隊(duì)列收到消息?:?{data=1631534511468,?正常隊(duì)列消息，最大長(zhǎng)度?5,?messageId=d63d2a4c-e7d3-4f00-a6ca-78e2d62d1d92}
死信隊(duì)列收到消息?:?{data=1631534585087,?正常隊(duì)列消息，最大長(zhǎng)度?5,?messageId=eed0c349-415b-43dc-aa79-c683122a1289}
死信隊(duì)列收到消息?:?{data=1631534588311,?正常隊(duì)列消息，最大長(zhǎng)度?5,?messageId=7a7bd152-f2fa-4a74-b9e6-943ac7cbb3d4}
死信隊(duì)列收到消息?:?{data=1631534608504,?正常隊(duì)列消息，最大長(zhǎng)度?5,?messageId=9de512a1-4ca4-4060-9096-27aba01c1687}

推薦閱讀：

SpringBoot實(shí)現(xiàn)人臉識(shí)別功能

JWT 登錄認(rèn)證及 token 自動(dòng)續(xù)期方案解讀

互聯(lián)網(wǎng)初中高級(jí)大廠面試題(9個(gè)G)

內(nèi)容包含Java基礎(chǔ)、JavaWeb、MySQL性能優(yōu)化、JVM、鎖、百萬(wàn)并發(fā)、消息隊(duì)列、高性能緩存、反射、Spring全家桶原理、微服務(wù)、Zookeeper......等技術(shù)棧！

?戳閱讀原文領(lǐng)??！? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ??朕已閱?