Redis中如何實現消息隊列和延時消息隊列

這篇文章將為大家詳細講解有關Redis中如何實現消息隊列和延時消息隊列，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

list的幾個命令

lpush (left push)

由隊列的左邊存放進去

rpush (right push)

由隊列的右邊存放進去

lpop  (left pop)

由隊列的左邊取出來

rpop (right pop)

由隊列的右邊取出來

以上的四個命令，可以讓 list 幫我們實現隊列 或者 棧，隊列的特性是先進先出，棧的特性是先進后出，

所以隊列的實現可以使用 lpush + rpop 或者 rpush + lpop，

棧的實現則是lpush + lpop 或者 rpush + rpop。

使用命令演示隊列

生產者發布消息

首先我們使用 rpush 對一個叫做notify-queue的隊列，增加五個元素，即 1 2 3 4 5，也就是作為生產者發布消息啦

消費者消費消息

既然生產者使用的是rpush，那么消費者就要用lpop，可以看下下圖，我們不停對notify-queue進行消息消費，并且是按照順序的，從1一直到5，按順序讀出，最終隊列中沒有消息了，彈出的則一直是空

空輪詢問題

在上面使用lpop消費消息時，可以看到，消息消費完后，我們每次再去pop時，讀到的都是一個空的信息，

上面是手動執行命令，但是如果是寫好的代碼程序不停的去pop數據(拉取數據)的話，會造成空輪詢(無用的讀取)，

既拉高了客戶端的CPU消耗，又拉高了redis的QPS，并且還是無用操作，這些無用操作可能會造成其他客戶端對redis的訪問變得響應緩慢。

解決方案A (休眠)

既然空輪詢會讓客戶端和redis的資源消耗都會變得較高，那么我們可以讓客戶端在收到空數據的時候，進行1s的休眠，1s后再進行數據拉取，這樣可以降低消耗

Thread.sleep(1000)

這個方案也是存在瑕疵的，即消息消費延遲性增大了，如果只有一個消費者的話，延遲就是1s，即空輪詢后，正好休眠了，但是這時候剛好有消息過來了，還是要等到1s醒來后才能消費，

如果有多個消費者的話，由于每個消費者的睡眠時間是岔開的，會降低一些延遲性，但是有沒有辦法更好的方法，可以做到幾乎 0 延遲？

解決方案B (阻塞讀)

redis中關于隊列取數據其實還有兩個命令，即阻塞讀取，

blpop (blocking left pop)

brpop (blocking right pop)

阻塞讀在隊列沒有數據的時候，會進入休眠狀態，一旦有消息來了以后，則立刻做出反應，讀取數據，因此使用 blpop/brpop 替換 lpop/rpop 則可以解決消息延遲性的問題，

繼續入隊3個屬性，6、7、8

使用 blpop 進行隊列的讀取，最后一個參數是阻塞讀的等待時間，如果超過這個時間還沒有消息，將會返回nil，此時可以繼續重復blpop操作,

阻塞讀的空閑連接自動斷開問題

客戶端使用阻塞讀時，如果阻塞的時間過長，服務一般會當成空閑連接，從而對其進行主動斷開，減少無用的連接占用資源，這個時候客戶端會拋出異常，

所以請注意，在客戶端使用阻塞讀的時候，要進行異常的捕獲，從而做出相應的處理，例如重試。

java客戶端實現消息隊列

思路和上述一樣，只是由命令行客戶端redis-cli變成了java語言，一個線程或多個線程進行 rpush 的發布，

另外一個或多個線程進行 blpop 消費，完成的代碼在：https://github.com/qiaomengnan16/redis-demo/tree/main/redis-queue

發布者

訂閱者

延時隊列的實現思路

延時隊列指的是，消息發送的一段時間后，再由消費者進行消費，而不是發送過去后，消費者就能立即讀取到，

zset的可以幫我們做到這個事情，首先zset可以通過score進行排序，score可以存一個時間戳，所以我們每次發布消息的時候，用當前時間戳加上延時的時間戳，

隨后消費者取消息的時候，通過截取zset的數據，取到已經滿足當前時間的消息（即取score小于等于當前時間戳的數據，score小于等于當前時間戳代表消息已經到時間了，如果大于的話，說明還得等一會才能消費）。

關鍵命令 zadd (發布者)，zrangebyscore（訂閱者），zrem (訂閱者消費完數據后刪除)

命令實現

我們使用zadd添加了4個數據，分別是1、2、3秒(偽說法，這里其實只是個score)后才能消費的數據，還有一個10秒后才能消費的kafka，

假如現在已經到了第三秒，我們取zset中大于等于1秒的和小于等于3秒的數據，因為這個區間的數據正好是我們可以消費的，可以看到，我們取出了符合條件的3條數據，

如果每次只能消費一個數據的話，可以加一個limit限制條件，可以看下圖取出了第一個可以消費的數據，redis

同時注意，和list的lpop/和blpop不同（它們彈出即自動刪除原始隊列里的該數據），

雖然獲取到了數據，但是如果不使用zrem進行刪除的話，這條數據還會被其他人讀到，因為他還一直存在zset中，

不過zrem可能會發生已經被別人搶先一步刪除(消費)的情況，所以代碼中還需要根據zrem的返回值是否大于0判斷，本次消息我們是否搶占成功，成功后再進行正確消費。

代碼實現

發布者

訂閱者

測試延遲效果

完整代碼地址：https://github.com/qiaomengnan16/redis-demo/tree/main/redis-delayed-queue

優化, 使用lua實現

上面實現的延遲隊列中，有一個問題，就是使用zrem判斷是否搶到這個數據時，很有可能沒有搶到，這樣繼續進行讀取，可能幾輪都搶不到，資源白白浪費了，所以可以通過lua腳本，來進行優化，

讓zrangebyscore 和 zrem成為一個原子化操作，這就可以避免多線程爭搶，搶不到的資源浪費了。

關于“Redis中如何實現消息隊列和延時消息隊列”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。