摘自：https://www.cnblogs.com/haoxinyue/p/6613706.html
場景一：物聯(lián)網(wǎng)系統(tǒng)經常會遇到向終端下發(fā)命令领突，如果命令一段時間沒有應答，就需要設置成超時。
場景二：訂單下單之后30分鐘后俏讹，如果用戶沒有付錢靠柑，則系統(tǒng)自動取消訂單具壮。
上述類似的需求是我們經常會遇見的問題梯轻。最常用的方法是定期輪訓數(shù)據(jù)庫，設置狀態(tài)输莺。在數(shù)據(jù)量小的時候并沒有什么大的問題，但是數(shù)據(jù)量一大輪訓數(shù)據(jù)庫的方式就會變得特別耗資源漾稀。當面對千萬級模闲、上億級數(shù)據(jù)量時，本身寫入的IO就比較高崭捍，導致長時間查詢或者根本就查不出來尸折，更別說分庫分表以后了。除此之外殷蛇，還有優(yōu)先級隊列实夹，基于優(yōu)先級隊列的JDK延遲隊列，時間輪等方式粒梦。但如果系統(tǒng)的架構中本身就有RabbitMQ的話亮航，那么選擇RabbitMQ來實現(xiàn)類似的功能也是一種選擇。
使用RabbitMQ來實現(xiàn)延遲任務必須先了解RabbitMQ的兩個概念：消息的TTL和死信Exchange匀们，通過這兩者的組合來實現(xiàn)上述需求缴淋。
消息的TTL（Time To Live）
消息的TTL就是消息的存活時間。RabbitMQ可以對隊列和消息分別設置TTL泄朴。對隊列設置就是隊列沒有消費者連著的保留時間重抖，也可以對每一個單獨的消息做單獨的設置。超過了這個時間祖灰，我們認為這個消息就死了钟沛，稱之為死信。如果隊列設置了局扶，消息也設置了恨统，那么會取小的。所以一個消息如果被路由到不同的隊列中三妈，這個消息死亡的時間有可能不一樣（不同的隊列設置）畜埋。這里單講單個消息的TTL，因為它才是實現(xiàn)延遲任務的關鍵畴蒲。
可以通過設置消息的expiration字段或者x-message-ttl屬性來設置時間由捎，兩者是一樣的效果。只是expiration字段是字符串參數(shù)饿凛，所以要寫個int類型的字符串：
byte[] messageBodyBytes = "Hello, world!".getBytes();
AMQP.BasicProperties properties =new AMQP.BasicProperties();
properties.setExpiration("60000");
channel.basicPublish("my-exchange", "routing-key", properties, messageBodyBytes);
當上面的消息扔到隊列中后狞玛，過了60秒软驰，如果沒有被消費，它就死了心肪。不會被消費者消費到锭亏。這個消息后面的，沒有“死掉”的消息對頂上來硬鞍，被消費者消費慧瘤。死信在隊列中并不會被刪除和釋放，它會被統(tǒng)計到隊列的消息數(shù)中去固该。單靠死信還不能實現(xiàn)延遲任務锅减，還要靠Dead Letter Exchange。
Dead Letter Exchanges
Exchage的概念在這里就不在贅述伐坏，可以從這里進行了解怔匣。一個消息在滿足如下條件下，會進死信路由桦沉，記住這里是路由而不是隊列每瞒，一個路由可以對應很多隊列。
1. 一個消息被Consumer拒收了纯露，并且reject方法的參數(shù)里requeue是false剿骨。也就是說不會被再次放在隊列里，被其他消費者使用埠褪。
2. 上面的消息的TTL到了浓利，消息過期了。
3. 隊列的長度限制滿了钞速。排在前面的消息會被丟棄或者扔到死信路由上贷掖。
Dead Letter Exchange其實就是一種普通的exchange，和創(chuàng)建其他exchange沒有兩樣玉工。只是在某一個設置Dead Letter Exchange的隊列中有消息過期了羽资，會自動觸發(fā)消息的轉發(fā)淘菩，發(fā)送到Dead Letter Exchange中去遵班。
實現(xiàn)延遲隊列
延遲任務通過消息的TTL和Dead Letter Exchange來實現(xiàn)。我們需要建立2個隊列潮改，一個用于發(fā)送消息狭郑，一個用于消息過期后的轉發(fā)目標隊列。
生產者輸出消息到Queue1汇在，并且這個消息是設置有有效時間的翰萨，比如60s。消息會在Queue1中等待60s糕殉，如果沒有消費者收掉的話亩鬼，它就是被轉發(fā)到Queue2殖告，Queue2有消費者，收到雳锋，處理延遲任務黄绩。
具體實現(xiàn)步驟如下：
第一步， 首先需要創(chuàng)建2個隊列玷过。Queue1和Queue2爽丹。Queue1是一個消息緩沖隊列，在這個隊列里面實現(xiàn)消息的過期轉發(fā)辛蚊。如下圖粤蝎，設置Dead letter exchange和Dead letter routing key。設置這兩個屬性就是當消息在這個隊列中expire后袋马，采用哪個路由發(fā)送初澎。這個dlx的exchange需要事先創(chuàng)建好，就是一個普通的exchange飞蛹。由于我們還需要向Queue1發(fā)送消息谤狡，那么還需要創(chuàng)建一個exchange，并且和Queue1綁定卧檐。例子中墓懂，exchange同樣取名：queue1。
我們還需要建一個Queue2霉囚，這個隊列用于消息在Queue1中過期后轉發(fā)的目標隊列捕仔。所以這個Queue2隊列建好以后，需要綁定Queue1設置的死信路由：dlx盈罐。完成Queue2的綁定以后榜跌，環(huán)境就搭建完成了。
第二步盅粪，實現(xiàn)消息的Producer钓葫。由于我們的目的是讓進入Queue1的消息過期，然后自動轉送到Queue2中票顾，所以發(fā)送的時候础浮，需要設置過期時間。
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();
? ? ? ? ? ? bytei = 10;
? ? ? ? ? ? while(i-- > 0) {? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? channel.basicPublish("queue1", "queue1",newAMQP.BasicProperties.Builder().expiration(String.valueOf(i * 1000)).build(),
? ? ? ? ? ? ? ? ? ? ? ? newbyte[] { i });
? ? ? ? ? ? }
上面的代碼我模擬了1-10號消息奠骄，消息的內容里面是1-10豆同。過期的時間是10-1秒。這里要注意含鳞，雖然10是第一個發(fā)送影锈，但是它過期的時間最長。
第三步，實現(xiàn)消息的Consumer鸭廷。Consumer就是延遲任務的具體實施者枣抱。由于具體的任務往往是一個比較耗時的任務，所以一般來說辆床，任務一般在異步線程中執(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) {
@OverridepublicvoidhandleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body)throws IOException {longdeliveryTag = envelope.getDeliveryTag();
? ? ? ? ? ? ? ? ? ? //do some work asyncSystem.out.println(body[0]);}
});
運行后如上面的程序，過了10s以后佛吓，消費者開始收到數(shù)據(jù)宵晚，但是它是一次性收到如下結果：
10、9 维雇、8 淤刃、7 、6吱型、5 逸贾、4 、3 津滞、2 铝侵、1
Consumer第一個收到的還是10。雖然10是第一個放進隊列触徐，但是它的過期時間最長咪鲜。所以由此可見，即使一個消息比在同一隊列中的其他消息提前過期撞鹉，提前過期的也不會優(yōu)先進入死信隊列疟丙，它們還是按照入庫的順序讓消費者消費。如果第一進去的消息過期時間是1小時鸟雏，那么死信隊列的消費者也許等1小時才能收到第一個消息享郊。參考官方文檔發(fā)現(xiàn)“Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered).”只有當過期的消息到了隊列的頂端（隊首），才會被真正的丟棄或者進入死信隊列孝鹊。
所以在考慮使用RabbitMQ來實現(xiàn)延遲任務隊列的時候炊琉，需要確保業(yè)務上每個任務的延遲時間是一致的。如果遇到不同的任務類型需要不同的延時的話又活，需要為每一種不同延遲時間的消息建立單獨的消息隊列苔咪。
