用 RabbitMQ 實(shí)現(xiàn)延遲隊(duì)列的兩種方案

1. 隊(duì)列或者消息的 TTL 配合 DLX

實(shí)現(xiàn)流程如下圖所示：

DLX 實(shí)現(xiàn)延遲隊(duì)列

生產(chǎn)者將消息發(fā)布到“緩沖隊(duì)列”（消息先發(fā)送到 exchange艇潭，這個(gè) exchange 綁定了一個(gè)設(shè)置了 message-ttl 的隊(duì)列拼窥，該隊(duì)列沒有任何消費(fèi)者，因此消息會(huì)在一定時(shí)間后超時(shí)）蹋凝。

消息超時(shí)后會(huì)被發(fā)布到“緩沖隊(duì)列”綁定的 DLX鲁纠，并被路由到對(duì)應(yīng)的“實(shí)際消費(fèi)隊(duì)列”，最終被消費(fèi)者消費(fèi)鳍寂。

隊(duì)列中消息的 TTL 可以設(shè)置在隊(duì)列上改含，有以下兩種方法：

• 設(shè)置隊(duì)列的 policy（message-ttl）。如果需要修改迄汛，可以直接在服務(wù)上用命令修改捍壤，無需改動(dòng)業(yè)務(wù)代碼骤视，無需刪除原有隊(duì)列。官方文檔推薦用這種方法設(shè)置 TTL鹃觉。具體命令如下：

rabbitmqctl set_policy expiry ".*" '{"expires":1800000}' --apply-to queues

• 設(shè)置 optional queue arguments 的 x-message-ttl 值专酗，由于屬性是聲明隊(duì)列時(shí)設(shè)置的，因此如果需要修改盗扇，則必須刪除原來的隊(duì)列而建新的隊(duì)列祷肯，并且修改相應(yīng)的業(yè)務(wù)代碼。這種方法可能適用于聲明 auto-delete 為 true 的隊(duì)列疗隶，由業(yè)務(wù)代碼聲明佑笋，可以靈活設(shè)置想要的 TTL。示例代碼如下：

// 聲明隊(duì)列時(shí)設(shè)置 x-message-ttl 屬性為 60000 毫秒
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 60000);
channel.queueDeclare("myqueue", false, false, false, args);

消息在發(fā)布時(shí)也可以單獨(dú)設(shè)置 TTL抽减。如果此時(shí)隊(duì)列也設(shè)置了 TTL允青，則取兩者較小的值起作用。示例代碼如下：

byte[] messageBodyBytes = "Hello, world!".getBytes();
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
    .expiration("60000")
    .build();
channel.basicPublish("my-exchange", "routing-key", properties, messageBodyBytes);

然而卵沉，根據(jù)官網(wǎng)文檔的描述：

Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered).

超時(shí)的消息只有到達(dá)了隊(duì)列的頭部才會(huì)真正被移除或變?yōu)樗佬拧?/p>

舉例來說颠锉，如果隊(duì)列設(shè)置了消息的超時(shí)時(shí)間為 10 秒，一個(gè)自身設(shè)置了超時(shí)時(shí)間為 5 秒的消息并不一定能保證其被加入隊(duì)列 5 秒后被移除史汗。因?yàn)檫@條消息之前可能已經(jīng)有很多消息插入隊(duì)列了琼掠，這些消息的超時(shí)時(shí)間都是 10 秒，那么這條消息就必須等前面的消息都被移除后才能被移除停撞。因此瓷蛙，即使單獨(dú)設(shè)置消息的超時(shí)時(shí)間，其行為也非常不可控戈毒，無法實(shí)現(xiàn)所謂任意時(shí)長(zhǎng)的延時(shí)效果艰猬。

消息超時(shí)變?yōu)樗佬藕螅瑫?huì)被發(fā)布到隊(duì)列配置的 DLX埋市，DLX 也是一個(gè)再普通不過的 exchange冠桃，發(fā)布到 DLX 的消息同樣會(huì)被路由到綁定到 DLX 的符合路由條件的隊(duì)列，這樣就實(shí)現(xiàn)了消息的延時(shí)消費(fèi)道宅。配置隊(duì)列的 DLX 同樣有兩種方式：

• 設(shè)置隊(duì)列的 policy（dead-letter-exchange）食听。具體命令如下：

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

• 設(shè)置 optional queue arguments 的 x-dead-letter-exchange 值。示例代碼如下：

// 聲明一個(gè) direct exchange
channel.exchangeDeclare("some.exchange.name", "direct");
// 聲明一個(gè)隊(duì)列污茵，將前面聲明的 exchange 設(shè)置為其 DLX
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "some.exchange.name");
channel.queueDeclare("myqueue", false, false, false, args);

緩沖隊(duì)列可以設(shè)置 x-dead-letter-routing-key 屬性樱报，如果不設(shè)置，消息進(jìn)入 DLX 后泞当，將仍用消息自身的 routing key 路由迹蛤。

限制：

• 由于延時(shí)消息實(shí)際存放在隊(duì)列中，因此延時(shí)不易設(shè)置過大，否則可能會(huì)導(dǎo)致大量消息積壓在隊(duì)列中笤受，占用 broker 資源穷缤。
• 不支持發(fā)送任意 TTL 的消息，TTL 必須與延時(shí)隊(duì)列綁定箩兽。

2. 使用 rabbitmq-delayed-message-exchange 插件

RabbitMQ v3.5.8 及以后版本支持該插件津肛，首先來看一下官方文檔上的一句話：

This plugin is considered to be experimental yet fairly stable and potential suitable for production use as long as the user is aware of its limitations.

該插件是實(shí)驗(yàn)性質(zhì)的，然而其也相當(dāng)穩(wěn)定并具有在生產(chǎn)中使用的潛質(zhì)汗贫，前提是使用者清楚它的局限性身坐。

rabbitmq-delayed-message-exchange 插件會(huì)將消息持久化到數(shù)據(jù)庫（Mnesia），再利用調(diào)度器調(diào)度消息的發(fā)布落包。因此延遲消息并不直接發(fā)布到任何 exchange部蛇，因此也不會(huì)直接保存到任何隊(duì)列里，而是先保存在磁盤上咐蝇，到了具體的需要發(fā)布的時(shí)間再發(fā)布涯鲁，因此消息在發(fā)布時(shí)會(huì)對(duì)于實(shí)際的發(fā)布時(shí)間有一定的延遲。

限制：

• 持久化的消息數(shù)據(jù)庫是單點(diǎn)的有序，一旦某個(gè)節(jié)點(diǎn)的丟失或者禁用該節(jié)點(diǎn)上的 rabbitmq-delayed-message-exchange 插件會(huì)導(dǎo)致這個(gè)節(jié)點(diǎn)上的所有延遲消息都丟失掉抹腿。
• 消息只會(huì)被嘗試發(fā)布一次，如果發(fā)布消息時(shí)沒有可以消費(fèi)的隊(duì)列存在旭寿，消息也不會(huì)被退回原發(fā)布者警绩，因?yàn)椴荒鼙ＷC此時(shí)原發(fā)布者還“活著”。
• 同時(shí)存在的延遲消息總數(shù)不能超過百萬級(jí)盅称。具體參見 issue 72肩祥。其實(shí)說白了，我們不應(yīng)該期待把 RabbitMQ 當(dāng)做數(shù)據(jù)庫來使用缩膝，每一個(gè)技術(shù)都有其適用條件混狠，如果有大量延遲消息需要持久化，完全可以考慮用其他方法解決問題疾层。

插件的安裝和使用方法參見官方文檔：RabbitMQ plugins将饺。

那么我們?cè)撊绾问褂眠@個(gè)插件實(shí)現(xiàn)延遲隊(duì)列呢？具體步驟如下：

• 使用 exchange 的擴(kuò)展屬性云芦，將 exchange 設(shè)置為”延遲“的俯逾，代碼如下：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("my-exchange", "x-delayed-message", true, false, args);

• 發(fā)布消息時(shí)設(shè)置消息 header 的 x-delayed 屬性贸桶，可以看到插件支持任意時(shí)長(zhǎng)的消息延時(shí)舅逸。代碼如下：

byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);
AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes);

參考資料

1. RabbitMQ 官方文檔
2. rabbitmq-delayed-message-exchange

實(shí)驗(yàn)代碼

所有實(shí)驗(yàn)代碼參見：github 源碼，注意皇筛，不同框架有不同的集成 RabbitMQ 的方案琉历，請(qǐng)參閱相關(guān)文檔，請(qǐng)不要在生產(chǎn)環(huán)境中直接使用實(shí)驗(yàn)代碼。