簡(jiǎn)介

本文介紹RabbitMQ的死信隊(duì)列和延遲隊(duì)列。

本內(nèi)容也是Java后端面試中常見的問題。

死信隊(duì)列

簡(jiǎn)介

DLX，全稱為Dead-Letter-Exchange,可以稱之為死信交換器，也有人稱之為死信郵箱。當(dāng)消息在一個(gè)隊(duì)列中變成死信（dead message)之后，它能被重新被發(fā)送到另一個(gè)交換器中，這個(gè)交換器就是DLX，綁定DLX的隊(duì)列就稱之為死信隊(duì)列。

以下幾種情況會(huì)導(dǎo)致消息變成死信：

• 消息被拒絕（Basic.Reject/Basic.Nack)，并且設(shè)置requeue參數(shù)為false;
• 消息過期；
• 隊(duì)列達(dá)到最大長(zhǎng)度。

DLX是一個(gè)正常的交換器，和一般的交換器沒有區(qū)別，它能在任何的隊(duì)列上被指定，實(shí)際上就是設(shè)置某個(gè)隊(duì)列的屬性。當(dāng)這個(gè)隊(duì)列中存在死信時(shí)，RabbitMQ就會(huì)自動(dòng)地將這個(gè)消息重新發(fā)布到設(shè)置的DLX上去，進(jìn)而被路由到另一個(gè)隊(duì)列，即死信隊(duì)列?？梢员O(jiān)聽這個(gè)隊(duì)列中的消息以進(jìn)行相應(yīng)的處理，這個(gè)特性與將消息的TTL設(shè)置為0配合使用可以彌補(bǔ)immediate參數(shù)的功能。

為隊(duì)列添加DLX的方法

法1：代碼方式

//創(chuàng)建 DLX: dlx_exchange
channel.exchangeDeclare("dlx_exchange", "direct" );
Map<String, Object> args = new HashMap<String, Object>;
args.put("x-dead-letter-exchange", "dlx_exchange");
//為隊(duì)列myqueue添加DLX
channel.queueDeclare("myqueue", false, false, false, args);

也可以為這個(gè)DLX指定路由鍵。（如果沒有特殊指定，則使用原隊(duì)列的路由鍵）

args.put("x-dead-letter-routing-key","dlx-routing-key"); 

法2：命令方式

rabbitmqctl set_policy DLX ".*" '{"dead-letter-exchange":"dlx_exchange"}' --apply-to queues

示例

代碼

channel.exchangeDeclare("exchange.dlx", "direct", true);
channel.exchangeDeclare("exchange.normal", "fanout", true);
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 10000);
args.put("x-dead-letter-exchange" , "exchange.dlx");
args.put("x-dead-letter-routing-key" , "routingkey");
channel.queueDeclare("queue.normal" , true, false, false, args);
channel.queueBind("queue.normal", "exchange.normal", "");
channel.queueDeclare("queue.dlx", true, false, false, null);
channel.queueBind("queue.dlx", "exchange.dlx" , "routingkey");
channel.basicPublish("exchange.normal" , "rk",
MessageProperties.PERSISTENT_TEXT_PLAIN, "dlx".getBytes());

這里創(chuàng)建了兩個(gè)交換器exchange.normal和exchange.dlx，分別綁定兩個(gè)隊(duì)列queue.normal和queue.dlx。

Web管理頁(yè)面結(jié)果

由下圖（圖1-1）的Web管理頁(yè)面可以看出，兩個(gè)隊(duì)列都被標(biāo)記了“D”，這個(gè)是durable的縮寫，即設(shè)置了隊(duì)列持久化。queue.normal這個(gè)隊(duì)列還配置了TTL、DLX和DLK，其中DLX指的是
x-dead-letter-routing-key這個(gè)屬性。 

圖1-1

案例分析

參考下圖（圖1-2），生產(chǎn)者首先發(fā)送一條攜帶路由鍵為“rk”的消息，然后經(jīng)過交換器exchange.normal順利地存儲(chǔ)到隊(duì)列queue.normal中。由于隊(duì)列queue.normal設(shè)置了過期時(shí)間為10s，在這10s內(nèi)沒有消費(fèi)者消費(fèi)這條消息，那么判定這條消息為過期。由于設(shè)置了DLX，過期之時(shí)，消息被丟給交換器exchange.dlx中，這時(shí)找到與exchange.dlx匹配的隊(duì)列queue.dlx，最后消息被存儲(chǔ)在queue.dk這個(gè)死信隊(duì)列中。 

圖1-2

對(duì)于RabbitMQ來說，DLX是一個(gè)非常有用的特性。它可以處理異常情況下，消息不能夠被消費(fèi)者正確消費(fèi)（消費(fèi)者調(diào)用了Basic.Nack或者Basic.Reject)而被置入死信隊(duì)列中的情況，后續(xù)分析程序可以通過消費(fèi)這個(gè)死信隊(duì)列中的內(nèi)容來分析當(dāng)時(shí)所遇到的異常情況，進(jìn)而可以改善和優(yōu)化系統(tǒng)。DLX配合TTL使用還可以實(shí)現(xiàn)延遲隊(duì)列的功能，詳細(xì)請(qǐng)看下一節(jié)。

延遲隊(duì)列

簡(jiǎn)介

延遲隊(duì)列用來存放延遲消息。延遲消息：指當(dāng)消息被發(fā)送以后，不想讓消費(fèi)者立刻拿到消息，而是等待特定時(shí)間后，消費(fèi)者才能拿到這個(gè)消息進(jìn)行消費(fèi)。

在AMQP協(xié)議中，或者RabbitMQ本身沒有直接支持延遲隊(duì)列的功能，但是有兩種方案來間接實(shí)現(xiàn)：

• 方案1：采用rabbitmq-delayed-message-exchange 插件實(shí)現(xiàn)。（RabbitMQ 3.6.x開始支持）
• 方案2：通過前面所介紹的DLX和TTL模擬出延遲隊(duì)列的功能。

在圖1-2中，不僅展示的是死信隊(duì)列的用法，也是延遲隊(duì)列的用法，對(duì)于queue.dlx這個(gè)死信隊(duì)列來說，同樣可以看作延遲隊(duì)列。假設(shè)一個(gè)應(yīng)用中需要將每條消息都設(shè)置為10秒的延遲，

生產(chǎn)者通過exchange.normal這個(gè)交換器將發(fā)送的消息存儲(chǔ)在queue.normal這個(gè)隊(duì)列中。消費(fèi)者訂閱的并非是queue.normal這個(gè)隊(duì)列，而是queue.dlx這個(gè)隊(duì)列。當(dāng)消息從queue.normal這個(gè)隊(duì)列中過期之后被存入queue.dlx這個(gè)隊(duì)列中，消費(fèi)者就恰巧消費(fèi)到了延遲10秒的這條消息。

在真實(shí)應(yīng)用中，對(duì)于延遲隊(duì)列可以根據(jù)延遲時(shí)間的長(zhǎng)短分為多個(gè)等級(jí)，一般分為5秒、10秒、30秒、1分鐘、5分鐘、10分鐘、30分鐘、1小時(shí)這幾個(gè)維度，當(dāng)然也可以再細(xì)化一下。

以下圖（圖2-1）為例進(jìn)行說明。為簡(jiǎn)化，只設(shè)置5秒、10秒、30秒、1分鐘這四個(gè)等級(jí)。根據(jù)需求的不同，生產(chǎn)者發(fā)送消息的時(shí)候通過設(shè)置不同的路由鍵，將消息發(fā)送到與交換器綁定的不同的隊(duì)列中。這里隊(duì)列也分別配置了DLX和相應(yīng)的死信隊(duì)列，當(dāng)相應(yīng)的消息過期時(shí)，就會(huì)轉(zhuǎn)存到相應(yīng)的死信隊(duì)列（即延遲隊(duì)列）中，這樣消費(fèi)者根據(jù)業(yè)務(wù)自身的情況，分別選擇不同延遲等級(jí)的延遲隊(duì)列進(jìn)行消費(fèi)。

圖2-1

使用場(chǎng)景

延遲隊(duì)列的使用場(chǎng)景有很多，比如：

用戶下訂單場(chǎng)景：用戶下單后有30分鐘的時(shí)間支付，若30分鐘內(nèi)沒有支付，則將這個(gè)訂單取消。

方案：用戶下單后將取消訂單的消息發(fā)送到延遲隊(duì)列，延遲時(shí)間設(shè)置為30分鐘。取消訂單這個(gè)消息的訂閱者程序在30分鐘后收到消息，判斷該訂單的狀態(tài)是否為已支付，若還沒支付，則將該訂單狀態(tài)設(shè)置為：已取消。

定時(shí)遙控場(chǎng)景：用戶想用手機(jī)遠(yuǎn)程遙控家里的智能設(shè)備在指定的時(shí)間工作。

方案：假設(shè)用戶想要的操作是：開啟熱水器。首先，將開啟熱水器這個(gè)消息發(fā)送到延遲隊(duì)列，延遲時(shí)間設(shè)置到用戶想要的時(shí)間到現(xiàn)在時(shí)間的差值。開啟熱水器這個(gè)消息的訂閱者程序在指定時(shí)間收到消息，再將指令推送到智能設(shè)備。

需要注意的是，延遲隊(duì)列的消息是不能取消的，解決方案是：在消費(fèi)消息的時(shí)候判斷這個(gè)消息對(duì)應(yīng)的業(yè)務(wù)的當(dāng)前狀態(tài)。例如：對(duì)于取消訂單來說，收到消息時(shí)，讀取這個(gè)消息所對(duì)應(yīng)的數(shù)據(jù)庫(kù)信息，如果已經(jīng)是已付款狀態(tài)了，就不進(jìn)行任何操作了，如果是未支付狀態(tài)，則改為已取消。

到此這篇關(guān)于詳解RabbitMQ中死信隊(duì)列和延遲隊(duì)列的使用詳解的文章就介紹到這了,更多相關(guān)RabbitMQ死信隊(duì)列 延遲隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論