所謂延時(shí)隊(duì)列就是延時(shí)的消息隊(duì)列，下面說(shuō)一下一些業(yè)務(wù)場(chǎng)景

實(shí)踐場(chǎng)景

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

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

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

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

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

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

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

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

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

>?rpush?queue?月伴飛魚(yú)1?月伴飛魚(yú)2?月伴飛魚(yú)3
(integer)?3
>?lpop?queue
"月伴飛魚(yú)1"
>?llen?queue
(integer)?2

問(wèn)題1:如果隊(duì)列空了

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

可是如果隊(duì)列空了，客戶端就會(huì)陷入 pop 的死循環(huán)，不停地 pop，沒(méi)有數(shù)據(jù)，接著再 pop，又沒(méi)有數(shù)據(jù)。這就是浪費(fèi)生命的空輪詢。空輪詢不但拉高了客戶端的 CPU，redis 的 QPS 也會(huì)被拉高，如果這樣空輪詢的客戶端有幾十來(lái)個(gè)，Redis 的慢查詢可能會(huì)顯著增多。

通常我們使用 sleep 來(lái)解決這個(gè)問(wèn)題，讓線程睡一會(huì)，睡個(gè) 1s 鐘就可以了。不但客戶端的 CPU 能降下來(lái)，Redis 的 QPS 也降下來(lái)了

問(wèn)題2:隊(duì)列延遲

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

有沒(méi)有什么辦法能顯著降低延遲呢？

那就是 blpop/brpop。

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

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

問(wèn)題3:空閑連接自動(dòng)斷開(kāi)

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

如果線程一直阻塞在哪里，Redis 的客戶端連接就成了閑置連接，閑置過(guò)久，服務(wù)器一般會(huì)主動(dòng)斷開(kāi)連接，減少閑置資源占用。這個(gè)時(shí)候 blpop/brpop 會(huì)拋出異常來(lái)。

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

分布式鎖沖突處理

假如客戶端在處理請(qǐng)求時(shí)加分布式鎖沒(méi)加成功怎么辦。

一般有 3 種策略來(lái)處理加鎖失敗：

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

2、sleep 一會(huì)再重試；

3、將請(qǐng)求轉(zhuǎn)移至延時(shí)隊(duì)列，過(guò)一會(huì)再試；

直接拋出特定類型的異常

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

sleep

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

延時(shí)隊(duì)列

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

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

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

private?Jedis?jedis;

public?void?redisDelayQueueTest()?{
????String?key?=?"delay_queue";

????//?實(shí)際開(kāi)發(fā)建議使用業(yè)務(wù)?ID?和隨機(jī)生成的唯一?ID?作為?value,?隨機(jī)生成的唯一?ID?可以保證消息的唯一性,?業(yè)務(wù)?ID?可以避免?value?攜帶的信息過(guò)多
????String?orderId1?=?UUID.randomUUID().toString();
????jedis.zadd(queueKey,?System.currentTimeMillis()?+?5000,?orderId1);

????String?orderId12?=?UUID.randomUUID().toString();
????jedis.zadd(queueKey,?System.currentTimeMillis()?+?5000,?orderId2);

????new?Thread()?{
????????@Override
????????public?void?run()?{
????????????while?(true)?{
????????????????Set?resultList;
????????????????//?只獲取第一條數(shù)據(jù),?只獲取不會(huì)移除數(shù)據(jù)
????????????????resultList?=?jedis.zrangebyscore(key,?System.currentTimeMillis(),?0,?1);
????????????????if?(resultList.size()?==?0)?{
????????????????????try?{
????????????????????????Thread.sleep(1000);
????????????????????}?catch?(InterruptedException?e)?{
????????????????????????e.printStackTrace();
????????????????????????break;
????????????????????}
????????????????}?else?{
????????????????????//?移除數(shù)據(jù)獲取到的數(shù)據(jù)
????????????????????if?(jedis.zrem(key,?resultList.get(0))?>?0)?{
????????????????????????String?orderId?=?resultList.get(0);
????????????????????????log.info("orderId?=?{}",?resultList.get(0));
????????????????????????this.handleMsg(orderId);
????????????????????}
????????????????}
????????????}
????????}
????}.start();
}

public?void?handleMsg(T?msg)?{
????System.out.println(msg);
}

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

1.T1, T2 和其他更多線程調(diào)用 zrangebyscore 獲取到了一條消息 A
2.T1 準(zhǔn)備開(kāi)始刪除消息 A, 由于是原子操作, T2 和其他更多線程等待 T1 執(zhí)行 zrem 刪除消息 A 后再執(zhí)行 zrem 刪除消息 A
3.T1 刪除了消息 A, 返回刪除成功標(biāo)記 1, 并對(duì)消息 A 進(jìn)行處理
4.T2 其他更多線程開(kāi)始 zrem 刪除消息 A, 由于消息 A 已經(jīng)被刪除, 所以所有的刪除均失敗, 放棄了對(duì)消息 A 的處理

同時(shí)，我們要注意一定要對(duì) handle_msg 進(jìn)行異常捕獲，避免因?yàn)閭€(gè)別任務(wù)處理問(wèn)題導(dǎo)致循環(huán)異常退出

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

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

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

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

String?luaScript?=?"local?resultArray?=?redis.call('zrangebyscore',?KEYS[1],?0,?ARGV[1],?'limit'?,?0,?1)\n"?+
????????"if?#resultArray?>?0?then\n"?+
????????"????if?redis.call('zrem',?KEYS[1],?resultArray[1])?>?0?then\n"?+
????????"????????return?resultArray[1]\n"?+
????????"????else\n"?+
????????"????????return?''\n"?+
????????"????end\n"?+
????????"else\n"?+
????????"????return?''\n"?+
????????"end";

jedis.eval(luaScript,?ScriptOutputType.VALUE,?new?String[]{key},?String.valueOf(System.currentTimeMillis()));

Redis延時(shí)隊(duì)列優(yōu)勢(shì)

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

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

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

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

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

Redis延時(shí)隊(duì)列劣勢(shì)

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

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

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

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

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

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

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

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

RDelayedQueue?delayedQueue?=?...
delayedQueue.destroy();

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

如果覺(jué)得不錯(cuò)，點(diǎn)個(gè)贊再走吧

參考

Redis Lua scripts debugger

Redis 深度歷險(xiǎn):核心原理與應(yīng)用實(shí)踐

Redis延時(shí)隊(duì)列，這次徹底給你整明白了