開發(fā)過程中通常會(huì)碰到這樣的需求：

• 淘寶訂單業(yè)務(wù)：下單后 30min 之內(nèi)沒有付款，就自動(dòng)取消訂單。
• 餓了嗎訂餐通知：下單成功后 60s 之后給用戶發(fā)送短信通知。
• 關(guān)閉空閑連接：服務(wù)器中有很多客戶端的連接，空閑一段時(shí)間之后需要關(guān)閉之。
• 緩存：緩存中的對(duì)象，超過了空閑時(shí)間，從緩存中移出。
• 任務(wù)超時(shí)處理：在網(wǎng)絡(luò)協(xié)議滑動(dòng)窗口請(qǐng)求應(yīng)答式交互時(shí)，處理超時(shí)未響應(yīng)的請(qǐng)求。
• 失敗重試機(jī)制：業(yè)務(wù)操作失敗后，間隔一定的時(shí)間進(jìn)行失敗重試。

這類業(yè)務(wù)的特點(diǎn)就是：需要延遲工作，需要進(jìn)行失敗重試。一種比較笨的方式是使用一個(gè)后臺(tái)線程，遍歷所有對(duì)象，挨個(gè)檢查。這種方法簡(jiǎn)單好用，但是對(duì)象數(shù)量過多時(shí)，可能存在性能問題，檢查間隔時(shí)間不好設(shè)置，間隔時(shí)間過大，影響精確度，過小則存在效率問題，而且做不到按超時(shí)的時(shí)間順序處理。

再比如常見的場(chǎng)景：

場(chǎng)景一：物聯(lián)網(wǎng)系統(tǒng)經(jīng)常會(huì)遇到向終端下發(fā)命令，如果命令一段時(shí)間沒有應(yīng)答，就需要設(shè)置成超時(shí)。

場(chǎng)景二：訂單下單之后30分鐘后，如果用戶沒有付錢，則系統(tǒng)自動(dòng)取消訂單。

上述類似的需求是我們經(jīng)常會(huì)遇見的問題。最常用的方法是定期輪訓(xùn)數(shù)據(jù)庫(kù)，設(shè)置狀態(tài)。在數(shù)據(jù)量小的時(shí)候并沒有什么大的問題，但是數(shù)據(jù)量一大輪訓(xùn)數(shù)據(jù)庫(kù)的方式就會(huì)變得特別耗資源。當(dāng)面對(duì)千萬級(jí)、上億級(jí)數(shù)據(jù)量時(shí)，本身寫入的IO就比較高，導(dǎo)致長(zhǎng)時(shí)間查詢或者根本就查不出來，更別說分庫(kù)分表以后了。除此之外，還有優(yōu)先級(jí)隊(duì)列，基于優(yōu)先級(jí)隊(duì)列的JDK延遲隊(duì)列，時(shí)間輪等方式。但如果系統(tǒng)的架構(gòu)中本身就有RabbitMQ的話，那么選擇RabbitMQ來實(shí)現(xiàn)類似的功能也是一種選擇。

使用RabbitMQ來實(shí)現(xiàn)延遲任務(wù)必須先了解RabbitMQ的兩個(gè)概念：消息的TTL和死信Exchange，通過這兩者的組合來實(shí)現(xiàn)上述需求。

消息的TTL（Time To Live）

消息的TTL就是消息的存活時(shí)間。RabbitMQ可以對(duì)隊(duì)列和消息分別設(shè)置TTL。對(duì)隊(duì)列設(shè)置就是隊(duì)列沒有消費(fèi)者連著的保留時(shí)間，也可以對(duì)每一個(gè)單獨(dú)的消息做單獨(dú)的設(shè)置。超過了這個(gè)時(shí)間，我們認(rèn)為這個(gè)消息就死了，稱之為死信。如果隊(duì)列設(shè)置了，消息也設(shè)置了，那么會(huì)取小的。所以一個(gè)消息如果被路由到不同的隊(duì)列中，這個(gè)消息死亡的時(shí)間有可能不一樣（不同的隊(duì)列設(shè)置）。這里單講單個(gè)消息的TTL，因?yàn)樗攀菍?shí)現(xiàn)延遲任務(wù)的關(guān)鍵。

可以通過設(shè)置消息的expiration字段或者x-message-ttl屬性來設(shè)置時(shí)間，兩者是一樣的效果。只是expiration字段是字符串參數(shù)，所以要寫個(gè)int類型的字符串：

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

當(dāng)上面的消息扔到隊(duì)列中后，過了60秒，如果沒有被消費(fèi)，它就死了。不會(huì)被消費(fèi)者消費(fèi)到。這個(gè)消息后面的，沒有“死掉”的消息對(duì)頂上來，被消費(fèi)者消費(fèi)。死信在隊(duì)列中并不會(huì)被刪除和釋放，它會(huì)被統(tǒng)計(jì)到隊(duì)列的消息數(shù)中去。單靠死信還不能實(shí)現(xiàn)延遲任務(wù)，還要靠Dead Letter Exchange。

Dead Letter Exchanges

Exchage的概念在這里就不在贅述，一個(gè)消息在滿足如下條件下，會(huì)進(jìn)死信路由，記住這里是路由而不是隊(duì)列，一個(gè)路由可以對(duì)應(yīng)很多隊(duì)列。

1. 一個(gè)消息被Consumer拒收了，并且reject方法的參數(shù)里requeue是false。也就是說不會(huì)被再次放在隊(duì)列里，被其他消費(fèi)者使用。

2. 上面的消息的TTL到了，消息過期了。

3. 隊(duì)列的長(zhǎng)度限制滿了。排在前面的消息會(huì)被丟棄或者扔到死信路由上。

Dead Letter Exchange其實(shí)就是一種普通的exchange，和創(chuàng)建其他exchange沒有兩樣。只是在某一個(gè)設(shè)置Dead Letter Exchange的隊(duì)列中有消息過期了，會(huì)自動(dòng)觸發(fā)消息的轉(zhuǎn)發(fā)，發(fā)送到Dead Letter Exchange中去。

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

延遲任務(wù)通過消息的TTL和Dead Letter Exchange來實(shí)現(xiàn)。我們需要建立2個(gè)隊(duì)列，一個(gè)用于發(fā)送消息，一個(gè)用于消息過期后的轉(zhuǎn)發(fā)目標(biāo)隊(duì)列。

生產(chǎn)者輸出消息到Queue1，并且這個(gè)消息是設(shè)置有有效時(shí)間的，比如60s。消息會(huì)在Queue1中等待60s，如果沒有消費(fèi)者收掉的話，它就是被轉(zhuǎn)發(fā)到Queue2，Queue2有消費(fèi)者，收到，處理延遲任務(wù)。

具體實(shí)現(xiàn)步驟如下：

第一步， 首先需要?jiǎng)?chuàng)建2個(gè)隊(duì)列。Queue1和Queue2。Queue1是一個(gè)消息緩沖隊(duì)列，在這個(gè)隊(duì)列里面實(shí)現(xiàn)消息的過期轉(zhuǎn)發(fā)。如下圖，設(shè)置Dead letter exchange和Dead letter routing key。設(shè)置這兩個(gè)屬性就是當(dāng)消息在這個(gè)隊(duì)列中expire后，采用哪個(gè)路由發(fā)送。這個(gè)dlx的exchange需要事先創(chuàng)建好，就是一個(gè)普通的exchange。由于我們還需要向Queue1發(fā)送消息，那么還需要?jiǎng)?chuàng)建一個(gè)exchange，并且和Queue1綁定。例子中，exchange同樣取名：queue1。

我們還需要建一個(gè)Queue2，這個(gè)隊(duì)列用于消息在Queue1中過期后轉(zhuǎn)發(fā)的目標(biāo)隊(duì)列。所以這個(gè)Queue2隊(duì)列建好以后，需要綁定Queue1設(shè)置的死信路由：dlx。完成Queue2的綁定以后，環(huán)境就搭建完成了。

第二步，實(shí)現(xiàn)消息的Producer。由于我們的目的是讓進(jìn)入Queue1的消息過期，然后自動(dòng)轉(zhuǎn)送到Queue2中，所以發(fā)送的時(shí)候，需要設(shè)置過期時(shí)間。

ConnectionFactory factory = new ConnectionFactory();
   factory.setUsername("bsp");
   factory.setPassword("123456");
   factory.setVirtualHost("/");
   factory.setHost("10.23.22.42");
   factory.setPort(5672);
   conn = factory.newConnection();
   channel = conn.createChannel();
   byte[] messageBodyBytes = "Hello, world!".getBytes();
   byte i = 10;
   while (i-- > 0) {    
    channel.basicPublish("queue1", "queue1", new AMQP.BasicProperties.Builder().expiration(String.valueOf(i * 1000)).build(),
      new byte[] { i });
   }

上面的代碼我模擬了1-10號(hào)消息，消息的內(nèi)容里面是1-10。過期的時(shí)間是10-1秒。這里要注意，雖然10是第一個(gè)發(fā)送，但是它過期的時(shí)間最長(zhǎng)。

第三步，實(shí)現(xiàn)消息的Consumer。Consumer就是延遲任務(wù)的具體實(shí)施者。由于具體的任務(wù)往往是一個(gè)比較耗時(shí)的任務(wù)，所以一般來說，任務(wù)一般在異步線程中執(zhí)行。

ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("bsp");
factory.setPassword("123456");
factory.setVirtualHost("/");
factory.setHost("10.23.22.42");
factory.setPort(5672);
conn = factory.newConnection();
channel = conn.createChannel();
channel.basicConsume("queue2", true, "consumer", new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {
long deliveryTag = envelope.getDeliveryTag();     
//do some work async
System.out.println(body[0]);
}
});

運(yùn)行后如上面的程序，過了10s以后，消費(fèi)者開始收到數(shù)據(jù)，但是它是一次性收到如下結(jié)果：

10、9 、8 、7 、6、5 、4 、3 、2 、1

Consumer第一個(gè)收到的還是10。雖然10是第一個(gè)放進(jìn)隊(duì)列，但是它的過期時(shí)間最長(zhǎng)。所以由此可見，即使一個(gè)消息比在同一隊(duì)列中的其他消息提前過期，提前過期的也不會(huì)優(yōu)先進(jìn)入死信隊(duì)列，它們還是按照入庫(kù)的順序讓消費(fèi)者消費(fèi)。如果第一進(jìn)去的消息過期時(shí)間是1小時(shí)，那么死信隊(duì)列的消費(fèi)者也許等1小時(shí)才能收到第一個(gè)消息。參考官方文檔發(fā)現(xiàn)“Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered).”只有當(dāng)過期的消息到了隊(duì)列的頂端（隊(duì)首），才會(huì)被真正的丟棄或者進(jìn)入死信隊(duì)列。

所以在考慮使用RabbitMQ來實(shí)現(xiàn)延遲任務(wù)隊(duì)列的時(shí)候，需要確保業(yè)務(wù)上每個(gè)任務(wù)的延遲時(shí)間是一致的。如果遇到不同的任務(wù)類型需要不同的延時(shí)的話，需要為每一種不同延遲時(shí)間的消息建立單獨(dú)的消息隊(duì)列。

總結(jié)

以上就是這篇文章的全部?jī)?nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，如果有疑問大家可以留言交流，謝謝大家對(duì)腳本之家的支持。

最新評(píng)論