Redis 延時隊列，一次性搞明白

所謂延時隊列就是延時的消息隊列，下面說一下一些業(yè)務(wù)場景

實(shí)踐場景

訂單支付失敗，每隔一段時間提醒用戶

用戶并發(fā)量的情況，可以延時2分鐘給用戶發(fā)短信

先來看看Redis實(shí)現(xiàn)普通的消息隊列

我們知道，對于專業(yè)的消息隊列中間件，如Kafka和RabbitMQ，消費(fèi)者在消費(fèi)消息之前要進(jìn)行一系列的繁瑣過程。

如RabbitMQ發(fā)消息之前要創(chuàng)建 Exchange，再創(chuàng)建 Queue，還要將 Queue 和 Exchange 通過某種規(guī)則綁定起來，發(fā)消息的時候要指定 routingkey，還要控制頭部信息

但是絕大 多數(shù)情況下，雖然我們的消息隊列只有一組消費(fèi)者，但還是需要經(jīng)歷上面一些過程。

有了 Redis，對于那些只有一組消費(fèi)者的消息隊列，使用 Redis 就可以非常輕松的搞定。Redis 的消息隊列不是專業(yè)的消息隊列，它沒有非常多的高級特性， 沒有 ack 保證，如果對消息的可靠性有著極致的追求，那么它就不適合使用

異步消息隊列基本實(shí)現(xiàn)

Redis 的 list(列表) 數(shù)據(jù)結(jié)構(gòu)常用來作為異步消息隊列使用，使用 rpush/lpush 操作入隊列， 使用 lpop 和 rpop 來出隊列

>rpushqueue月伴飛魚1月伴飛魚2月伴飛魚3 (integer)3 >lpopqueue "月伴飛魚1" >llenqueue (integer)2

問題1:如果隊列空了

客戶端是通過隊列的 pop 操作來獲取消息，然后進(jìn)行處理。處理完了再接著獲取消息， 再進(jìn)行處理。如此循環(huán)往復(fù)，這便是作為隊列消費(fèi)者的客戶端的生命周期。

可是如果隊列空了，客戶端就會陷入 pop 的死循環(huán)，不停地 pop，沒有數(shù)據(jù)，接著再 pop， 又沒有數(shù)據(jù)。這就是浪費(fèi)生命的空輪詢?？蛰喸儾坏吡丝蛻舳说?CPU，redis 的 QPS 也 會被拉高，如果這樣空輪詢的客戶端有幾十來個，Redis 的慢查詢可能會顯著增多。

通常我們使用 sleep 來解決這個問題，讓線程睡一會，睡個 1s 鐘就可以了。不但客戶端 的 CPU 能降下來，Redis 的 QPS 也降下來了

問題2:隊列延遲

用上面睡眠的辦法可以解決問題。同時如果只有 1 個消費(fèi)者，那么這個延遲就是 1s。如果有多個消費(fèi)者，這個延遲會有所下降，因 為每個消費(fèi)者的睡覺時間是岔開來的。

有沒有什么辦法能顯著降低延遲呢？

那就是 blpop/brpop。

這兩個指令的前綴字符 b 代表的是 blocking，也就是阻塞讀。

阻塞讀在隊列沒有數(shù)據(jù)的時候，會立即進(jìn)入休眠狀態(tài)，一旦數(shù)據(jù)到來，則立刻醒過來。消 息的延遲幾乎為零。用 blpop/brpop 替代前面的 lpop/rpop，就完美解決了上面的問題。

問題3:空閑連接自動斷開

其實(shí)他還有個問題需要解決—— 空閑連接的問題。

如果線程一直阻塞在哪里，Redis 的客戶端連接就成了閑置連接，閑置過久，服務(wù)器一般 會主動斷開連接，減少閑置資源占用。這個時候 blpop/brpop 會拋出異常來。

所以編寫客戶端消費(fèi)者的時候要小心，注意捕獲異常，還要重試。

分布式鎖沖突處理

假如客戶端在處理請求時加分布式鎖沒加成功怎么辦。

一般有 3 種策略來處理加鎖失?。?/p>

1、直接拋出異常，通知用戶稍后重試；

2、sleep 一會再重試；

3、將請求轉(zhuǎn)移至延時隊列，過一會再試；

直接拋出特定類型的異常

這種方式比較適合由用戶直接發(fā)起的請求，用戶看到錯誤對話框后，會先閱讀對話框的內(nèi) 容，再點(diǎn)擊重試，這樣就可以起到人工延時的效果。如果考慮到用戶體驗，可以由前端的代碼 替代用戶自己來進(jìn)行延時重試控制。它本質(zhì)上是對當(dāng)前請求的放棄，由用戶決定是否重新發(fā)起 新的請求。

sleep

sleep 會阻塞當(dāng)前的消息處理線程，會導(dǎo)致隊列的后續(xù)消息處理出現(xiàn)延遲。如果碰撞的比 較頻繁或者隊列里消息比較多，sleep 可能并不合適。如果因為個別死鎖的 key 導(dǎo)致加鎖不成 功，線程會徹底堵死，導(dǎo)致后續(xù)消息永遠(yuǎn)得不到及時處理。

延時隊列

這種方式比較適合異步消息處理，將當(dāng)前沖突的請求扔到另一個隊列延后處理以避開沖突。

延時隊列的實(shí)現(xiàn)

我們可以使用 zset這個命令，用設(shè)置好的時間戳作為score進(jìn)行排序，使用zadd score1 value1 ....命令就可以一直往內(nèi)存中生產(chǎn)消息。再利用 zrangebysocre 查詢符合條件的所有待處理的任務(wù)，通過循環(huán)執(zhí)行隊列任務(wù)即可。也可以通過zrangebyscore key min max withscores limit 0 1查詢最早的一條任務(wù)，來進(jìn)行消費(fèi)

privateJedisjedis; publicvoidredisDelayQueueTest(){ Stringkey="delay_queue"; //實(shí)際開發(fā)建議使用業(yè)務(wù)ID和隨機(jī)生成的唯一ID作為value,隨機(jī)生成的唯一ID可以保證消息的唯一性,業(yè)務(wù)ID可以避免value攜帶的信息過多 StringorderId1=UUID.randomUUID().toString(); jedis.zadd(queueKey,System.currentTimeMillis()+5000,orderId1); StringorderId12=UUID.randomUUID().toString(); jedis.zadd(queueKey,System.currentTimeMillis()+5000,orderId2); newThread(){ @Override publicvoidrun(){ while(true){ SetresultList; //只獲取第一條數(shù)據(jù),只獲取不會移除數(shù)據(jù) resultList=jedis.zrangebyscore(key,System.currentTimeMillis(),0,1); if(resultList.size()==0){ try{ Thread.sleep(1000); }catch(InterruptedExceptione){ e.printStackTrace(); break; } }else{ //移除數(shù)據(jù)獲取到的數(shù)據(jù) if(jedis.zrem(key,resultList.get(0))>0){ StringorderId=resultList.get(0); log.info("orderId={}",resultList.get(0)); this.handleMsg(orderId); } } } } }.start(); } publicvoidhandleMsg(Tmsg){ System.out.println(msg); }

上面的實(shí)現(xiàn), 在多線程邏輯上也是沒有問題的, 假設(shè)有兩個線程 T1, T2和其他更多線程, 處理邏輯如下, 保證了多線程情況下只有一個線程處理了對應(yīng)的消息:

1.T1, T2 和其他更多線程調(diào)用 zrangebyscore 獲取到了一條消息 A

2.T1 準(zhǔn)備開始刪除消息 A, 由于是原子操作, T2 和其他更多線程等待 T1 執(zhí)行 zrem 刪除消息 A 后再執(zhí)行 zrem 刪除消息 A

3.T1 刪除了消息 A, 返回刪除成功標(biāo)記 1, 并對消息 A 進(jìn)行處理

4.T2 其他更多線程開始 zrem 刪除消息 A, 由于消息 A 已經(jīng)被刪除, 所以所有的刪除均失敗, 放棄了對消息 A 的處理

同時，我們要注意一定要對handle_msg進(jìn)行異常捕獲，避免因為個別任務(wù)處理問題導(dǎo)致循環(huán)異常退 出

進(jìn)一步優(yōu)化

上面的算法中同一個任務(wù)可能會被多個進(jìn)程取到之后再使用 zrem 進(jìn)行爭搶，那些沒搶到 的進(jìn)程都是白取了一次任務(wù)，這是浪費(fèi)。可以考慮使用 lua scripting 來優(yōu)化一下這個邏輯，將 zrangebyscore 和 zrem 一同挪到服務(wù)器端進(jìn)行原子化操作，這樣多個進(jìn)程之間爭搶任務(wù)時就不 會出現(xiàn)這種浪費(fèi)了

使用調(diào)用Lua腳本進(jìn)一步優(yōu)化

Lua 腳本, 如果有超時的消息, 就刪除, 并返回這條消息, 否則返回空字符串:

StringluaScript="localresultArray=redis.call('zrangebyscore',KEYS[1],0,ARGV[1],'limit',0,1) "+ "if#resultArray>0then "+ "ifredis.call('zrem',KEYS[1],resultArray[1])>0then "+ "returnresultArray[1] "+ "else "+ "return'' "+ "end "+ "else "+ "return'' "+ "end"; jedis.eval(luaScript,ScriptOutputType.VALUE,newString[]{key},String.valueOf(System.currentTimeMillis()));

Redis延時隊列優(yōu)勢

Redis用來進(jìn)行實(shí)現(xiàn)延時隊列是具有這些優(yōu)勢的：

1.Redis zset支持高性能的 score 排序。

2.Redis是在內(nèi)存上進(jìn)行操作的，速度非?？?。

3.Redis可以搭建集群，當(dāng)消息很多時候，我們可以用集群來提高消息處理的速度，提高可用性。

4.Redis具有持久化機(jī)制，當(dāng)出現(xiàn)故障的時候，可以通過AOF和RDB方式來對數(shù)據(jù)進(jìn)行恢復(fù)，保證了數(shù)據(jù)的可靠性

Redis延時隊列劣勢

使用 Redis 實(shí)現(xiàn)的延時消息隊列也存在數(shù)據(jù)持久化, 消息可靠性的問題

沒有重試機(jī)制 - 處理消息出現(xiàn)異常沒有重試機(jī)制, 這些需要自己去實(shí)現(xiàn), 包括重試次數(shù)的實(shí)現(xiàn)等

沒有 ACK 機(jī)制 - 例如在獲取消息并已經(jīng)刪除了消息情況下, 正在處理消息的時候客戶端崩潰了, 這條正在處理的這些消息就會丟失, MQ 是需要明確的返回一個值給 MQ 才會認(rèn)為這個消息是被正確的消費(fèi)了

如果對消息可靠性要求較高, 推薦使用 MQ 來實(shí)現(xiàn)

Redission實(shí)現(xiàn)延時隊列

基于Redis的Redisson分布式延遲隊列結(jié)構(gòu)的RDelayedQueue Java對象在實(shí)現(xiàn)了RQueue接口的基礎(chǔ)上提供了向隊列按要求延遲添加項目的功能。該功能可以用來實(shí)現(xiàn)消息傳送延遲按幾何增長或幾何衰減的發(fā)送策略

RQueuedistinationQueue=... RDelayedQueuedelayedQueue=getDelayedQueue(distinationQueue); //10秒鐘以后將消息發(fā)送到指定隊列 delayedQueue.offer("msg1",10,TimeUnit.SECONDS); //一分鐘以后將消息發(fā)送到指定隊列 delayedQueue.offer("msg2",1,TimeUnit.MINUTES);

在該對象不再需要的情況下，應(yīng)該主動銷毀。僅在相關(guān)的Redisson對象也需要關(guān)閉的時候可以不用主動銷毀。

RDelayedQueuedelayedQueue=... delayedQueue.destroy();

是不是很方便...............

責(zé)任編輯：xj

原文標(biāo)題：Redis 延時隊列，這次徹底給你整明白了

文章出處：【微信公眾號：數(shù)據(jù)分析與開發(fā)】歡迎添加關(guān)注！文章轉(zhuǎn)載請注明出處。