Redis實現(xiàn)消息隊列的4種方案

Redis作為內(nèi)存中的數(shù)據(jù)結(jié)構(gòu)存儲，常用作數(shù)據(jù)庫、緩存和消息代理。它支持數(shù)據(jù)結(jié)構(gòu)，如 字符串，散列，列表，集合，帶有范圍查詢的排序集（sorted sets），位圖（bitmaps），超級日志（hyperloglogs），具有半徑查詢和流的地理空間索引。Redis具有內(nèi)置復(fù)制，Lua腳本，LRU驅(qū)逐，事務(wù)和不同級別的磁盤持久性，并通過Redis Sentinel和Redis Cluster自動分區(qū)。

為了實現(xiàn)其出色的性能，Redis使用內(nèi)存數(shù)據(jù)集（in-memory dataset）。

MQ應(yīng)用有很多，比如ActiveMQ,RabbitMQ,Kafka等，但是也可以基于redis來實現(xiàn)，可以降低系統(tǒng)的維護成本和實現(xiàn)復(fù)雜度，本篇介紹redis中實現(xiàn)消息隊列的幾種方案。

? ?1. 基于List的 LPUSH+BRPOP 的實現(xiàn)
? ?2. PUB/SUB，訂閱/發(fā)布模式
? ?3. 基于Sorted-Set的實現(xiàn)
? ?4. 基于Stream類型的實現(xiàn)

基于異步消息隊列List lpush-brpop(rpush-blpop)

使用rpush和lpush操作入隊列，lpop和rpop操作出隊列。

List支持多個生產(chǎn)者和消費者并發(fā)進出消息，每個消費者拿到都是不同的列表元素。

但是當隊列為空時，lpop和rpop會一直空輪訓(xùn)，消耗資源；所以引入阻塞讀blpop和brpop（b代表blocking），阻塞讀在隊列沒有數(shù)據(jù)的時候進入休眠狀態(tài)，?一旦數(shù)據(jù)到來則立刻醒過來，消息延遲幾乎為零。

?注意

你以為上面的方案很完美？還有個問題需要解決：空閑連接的問題。
?

如果線程一直阻塞在那里，Redis客戶端的連接就成了閑置連接，閑置過久，服務(wù)器一般會主動斷開連接，減少閑置資源占用，這個時候blpop和brpop或拋出異常，所以在編寫客戶端消費者的時候要小心，如果捕獲到異常，還有重試。

?

缺點：

• 做消費者確認ACK麻煩，不能保證消費者消費消息后是否成功處理的問題（宕機或處理異常等），通常需要維護一個Pending列表，保證消息處理確認。

• 不能做廣播模式，如pub/sub，消息發(fā)布/訂閱模型

• 不能重復(fù)消費，一旦消費就會被刪除

• 不支持分組消費

PUB/SUB,訂閱/發(fā)布模式

• SUBSCRIBE，用于訂閱信道

• PUBLISH，向信道發(fā)送消息

• UNSUBSCRIBE，取消訂閱

此模式允許生產(chǎn)者只生產(chǎn)一次消息，由中間件負責(zé)將消息復(fù)制到多個消息隊列，每個消息隊列由對應(yīng)的消費組消費。

優(yōu)點

• 典型的廣播模式，一個消息可以發(fā)布到多個消費者

• 多信道訂閱，消費者可以同時訂閱多個信道，從而接收多類消息

• 消息即時發(fā)送，消息不用等待消費者讀取，消費者會自動接收到信道發(fā)布的消息

缺點

• 消息一旦發(fā)布，不能接收。換句話就是發(fā)布時若客戶端不在線，則消息丟失，不能尋回

• 不能保證每個消費者接收的時間是一致的

• 若消費者客戶端出現(xiàn)消息積壓，到一定程度，會被強制斷開，導(dǎo)致消息意外丟失。通常發(fā)生在消息的生產(chǎn)遠大于消費速度時

可見，Pub/Sub 模式不適合做消息存儲，消息積壓類的業(yè)務(wù)，而是擅長處理廣播，即時通訊，即時反饋的業(yè)務(wù)。

基于Sorted-Set的實現(xiàn)

Sortes Set(有序列表)，類似于java的SortedSet和HashMap的結(jié)合體，一方面她是一個set，保證內(nèi)部value的唯一性，另一方面它可以給每個value賦予一個score，代表這個value的排序權(quán)重。內(nèi)部實現(xiàn)是“跳躍表”。

有序集合的方案是在自己確定消息順I(yè)D時比較常用，使用集合成員的Score來作為消息ID，保證順序，還可以保證消息ID的單調(diào)遞增。通常可以使用時間戳+序號的方案。確保了消息ID的單調(diào)遞增，利用SortedSet的依據(jù)

Score排序的特征，就可以制作一個有序的消息隊列了。

優(yōu)點

就是可以自定義消息ID，在消息ID有意義時，比較重要。
?

缺點
?

缺點也明顯，不允許重復(fù)消息（因為是集合），同時消息ID確定有錯誤會導(dǎo)致消息的順序出錯。

基于Stream類型的實現(xiàn)

Stream為redis 5.0后新增的數(shù)據(jù)結(jié)構(gòu)。支持多播的可持久化消息隊列，實現(xiàn)借鑒了Kafka設(shè)計。

Redis Stream的結(jié)構(gòu)如上圖所示，它有一個消息鏈表，將所有加入的消息都串起來，每個消息都有一個唯一的ID和對應(yīng)的內(nèi)容。消息是持久化的，Redis重啟后，內(nèi)容還在。

每個Stream都有唯一的名稱，它就是Redis的key，在我們首次使用xadd指令追加消息時自動創(chuàng)建。

每個Stream都可以掛多個消費組，每個消費組會有個游標last_delivered_id在Stream數(shù)組之上往前移動，表示當前消費組已經(jīng)消費到哪條消息了。每個消費組都有一個Stream內(nèi)唯一的名稱，消費組不會自動創(chuàng)建，它需要單獨的指令xgroup create進行創(chuàng)建，需要指定從Stream的某個消息ID開始消費，這個ID用來初始化last_delivered_id變量。

每個消費組(Consumer Group)的狀態(tài)都是獨立的，相互不受影響。也就是說同一份Stream內(nèi)部的消息會被每個消費組都消費到。

同一個消費組(Consumer Group)可以掛接多個消費者(Consumer)，這些消費者之間是競爭關(guān)系，任意一個消費者讀取了消息都會使游標last_delivered_id往前移動。每個消費者者有一個組內(nèi)唯一名稱。

消費者(Consumer)內(nèi)部會有個狀態(tài)變量pending_ids，它記錄了當前已經(jīng)被客戶端讀取的消息，但是還沒有ack。如果客戶端沒有ack，這個變量里面的消息ID會越來越多，一旦某個消息被ack，它就開始減少。這個pending_ids變量在Redis官方被稱之為PEL，也就是Pending Entries List，這是一個很核心的數(shù)據(jù)結(jié)構(gòu)，它用來確保客戶端至少消費了消息一次，而不會在網(wǎng)絡(luò)傳輸?shù)闹型緛G失了沒處理。

增刪改查

127.0.0.1:6379> xlen?codehole
(integer) 5
127.0.0.1:6379> xadd?codehole?maxlen?3 * name?xiaorui?age?1
1527855160273-0
127.0.0.1:6379> xlen?codehole
(integer) 3