Redis5新特性Streams作消息隊列
前言
Redis 5 新特性中，Streams 數據結構的引入，可以說它是在本次迭代中最大特性。它使本次 5.x 版本迭代中，Redis 作為消息隊列使用時，得到更完善，更強大的原生支持，其中尤為明顯的是持久化消息隊列。同時，stream 借鑒了 kafka 的消費組模型概念和設計，使消費消息處理上更加高效快速。本文就 Streams 數據結構中常用 API 進行分析。
準備
本文所使用 Redis 版本為 5.0.5 。如果使用更早的 5.x 版本，有些 API 使用效果，與本文中描述略有不同。
添加消息
Streams 添加數據使用 XADD 指令進行添加，消息中的數據以 K-V 鍵值對的形式進行操作。一條消息可以存在多個鍵值對，添加命令格式：
- XADD key ID field string [field string ...]
其中 key 為 Streams 的名稱，ID 為消息的唯一標志，不可重復，field string 就為鍵值對。下面我們就添加以 person 為名稱的流，進行操作。
- XADD person * name ytao des https://ytao.top
上面添加案例中，ID 使用 * 號復制，這里代表著服務端自動生成 Id，添加后返回數據 "1578238486193-0"
這里自動生成的 Id 格式為 <millisecondstime>-<sequencenumber> Id 是由兩部分組成：
- millisecondsTime 為當前服務器時間毫秒時間戳。
- sequenceNumber 當前序列號，取值來源于當前毫秒內，生成消息的順序，默認從 0 開始加 1 遞增。
比如：1578238486193-3 表示在 1578238486193 毫秒的時間戳時，添加的第 4 條消息。
除了服務端自動生成 Id 方式外，也支持指定 Id 的生成，但是指定 Id 有以下條件限制：
- Id 中的前后部分必須為數字。
- 最小 Id 為 0-1，不能為 0-0，但是 2-0，3-0 .... 是被允許的。
- 添加的消息，Id 的前半部分不能比存在 Id 最大的值小，Id 后半部分不能比存在前半部分相同的最大后半部分小。
否則，當不滿足上述條件時，添加后會拋出異常：
- (error) ERR The ID specified in XADD is equal or smaller than the target stream top item
實際上，當添加一條消息時，會進行兩部操作。第一步，先判斷如果不存在 Streams，則創建 Streams 的名稱，再添加消息到 Streams 中。即使添加消息時，由于 Id 異常，也可以在 Redis 中存在以當前 Streams 的名稱。 Streams 中 Id 也可作為指針使用，因為它是一個有序的標記。
生產中，如果這樣使用添加消息，會存在一個問題，那就是消息數量太大時，會使服務宕機。這里 Streams 的設計初期也有考慮到這個問題，那就是可以指定 Streams 的容量。如果容量操作這個設定的值，就會對調舊的消息。在添加消息時，設置 MAXLEN 參數。
- XADD person MAXLEN 5 * name ytao des https://ytao.top
這樣就指定該了 Streams 中的容量為 5 條消息。也可使用 XTRIM 截取消息，從小到大剔除多余的消息：
- XTRIM person MAXLEN 8
消息數量
查看消息數量使用 XLEN 指令進行操作。
- XLEN key
例：查看 person 流中的消息數量：
- > XLEN person
- (integer) 5
查詢消息
查詢 Streams 中的消息使用 XRANGE 和 XREVRANGE 指令。
XRANGE
查詢數據時，可以按照指定 Id 范圍進行查詢，XRANGE 查詢指令格式：
- XRANGE key start end [COUNT count]
參數說明：
- key 為 Streams 的名稱
- start 為范圍查詢開始 Id，包含本 Id。
- start 為范圍查詢結束 Id，包含本 Id。
- Count 為查詢返回最大的消息數量，非必填。
這里 start 和 end 有-和+兩個非指定值，他們分別表示無窮小和無窮大，所以當使用這個兩個值時，會查詢出全部的消息。
- > XRANGE person - +
- 1) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
- 2) 1) "0-2"
- 2) 1) "name"
- 2) "luffy"
- 3) "des"
- 4) "valiant!"
- 3) 1) "2-0"
- 2) 1) "name"
- 2) "gaga"
- 3) "des"
- 4) "fishion!"
上面查詢的消息數據，可以看到是按照先進先出的順序查詢出來的。
使用 COUNT 指定查詢返回的數量：
- # 查詢所有的消息，并且返回一條數據
- > XRANGE person - + COUNT 1
- 1) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
在范圍查詢中，Id 的后半部分可省略，后半部分中的數據會全部查詢到。
XREVRANGE
XREVRANGE 的查詢和 XRANGE 指令中的使用類似，但查詢的 start 和 end 參數順序進行了調換：
- XREVRANGE key end start [COUNT count]
使用案例：
- > XREVRANGE person + -
- 1) 1) "2-0"
- 2) 1) "name"
- 2) "gaga"
- 3) "des"
- 4) "fishion!"
- 2) 1) "0-2"
- 2) 1) "name"
- 2) "luffy"
- 3) "des"
- 4) "valiant!"
- 3) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
查詢后的結果與 XRANGE 的結果順序剛好相反，其他都一樣，這兩個指令可進行消息的升序和降序的返回。
刪除消息
刪除消息使用 XDEL 指令操作，只需指定將要刪除的 Streams 名稱和 Id 即可，支持一次刪除多個消息 。
- XDEL key ID [ID ...]
刪除案例：
- # 查詢所有消息
- > XRANGE person - +
- 1) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
- 2) 1) "0-2"
- 2) 1) "name"
- 2) "luffy"
- 3) "des"
- 4) "valiant!"
- 3) 1) "2-0"
- 2) 1) "name"
- 2) "gaga"
- 3) "des"
- 4) "fishion!"
- # 刪除消息
- > XDEL person 2-0
- (integer) 1
- # 再次查詢刪除后的所有消息
- > XRANGE person - +
- 1) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
- 2) 1) "0-2"
- 2) 1) "name"
- 2) "luffy"
- 3) "des"
- 4) "valiant!"
- # 查詢刪除后的長度
- > XLEN person
- (integer) 2
從上面可以看到，刪除消息后，長度也會減少相應的數量。
消費消息
在 Redis 的 PUB/SUB 中，我們是通過訂閱來消費消息，在 Streams 數據結構中，同樣也能實現同等功能，當沒有新的消息時，可進行阻塞等待。不僅支持單獨消費，而且還可以支持群組消費。
單獨消費
單獨消費使用 XREAD 指令。可以看到，下面命令中，STREAMS，key, 以及 ID 為必填項。ID 表示將要讀取大于該 ID 的消息。當 ID 值使用 $ 賦予時，表示已存在消息的最大 Id 值。
- XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
上面的 COUNT 參數用來指定讀取的最大數量，與 XRANGE 的用法一樣。
- > XREAD COUNT 1 STREAMS person 0
- 1) 1) "person"
- 2) 1) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
- > XREAD COUNT 2 STREAMS person 0
- 1) 1) "person"
- 2) 1) 1) "0-1"
- 2) 1) "name"
- 2) "ytao"
- 3) "des"
- 4) "https://ytao.top"
- 2) 1) "0-2"
- 2) 1) "name"
- 2) "luffy"
- 3) "des"
- 4) "valiant!"
在 XREAD 里面還有個 BLOCK 參數，這個是用來阻塞訂閱消息的，BLOCK 攜帶的參數為阻塞時間，單位為毫秒，如果在這個時間內沒有新的消息消費，那么就會釋放該阻塞。當這里的時間指定為 0 時，會一直阻塞，直到有新的消息來消費到。
- # 窗口 1 開啟阻塞，等待新消息的到來
- > XREAD BLOCK 0 STREAMS person $
- # 另開一個連接窗口 2，添加一條新的消息
- > XADD person 2-2 name tao des coder
- "2-2"
- # 窗口 1，獲取到有新的消息來消費，并且帶有阻塞的時間
- > XREAD BLOCK 0 STREAMS person $
- 1) 1) "person"
- 2) 1) 1) "2-2"
- 2) 1) "name"
- 2) "tao"
- 3) "des"
- 4) "coder"
- (60.81s)
當使用 XREAD 進行順序消費時，需要額外記錄下讀取到位置的 Id，方便下次繼續消費。
群組消費
群組消費的主要目的也就是為了分流消息給不同的客戶端處理，以更高效的速率處理消息。為達到這一肝功能需求，我們需要做三件事：創建群組，群組讀取消息，向服務端確認消息以處理。
群組操作
操作群組使用 XGROUP 指令：
- XGROUP [CREATE key groupname id-or-$] [SETID key id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
上面命令中，包含操作有：
- CREATE 創建消費組。
- SETID 修改下一個處理消息的 Id。
- DESTROY 銷毀消費組。
- DELCONSUMER 刪除消費組中指定的消費者。
我們當前需要使用的是創建消費組：
- # 以當前存在的最大 Id 作為消費起始
- > XGROUP CREATE person group1 $
- OK
群組讀取消息
群組讀取使用 XREADGROUP 指令，COUNT和BLOCK的使用類似 XREAD 的操作，只是多了個群組和消費者的指定：
- XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
由于群組消費和單獨消費類似，這里只進行個阻塞分析，這里 Id 也有個特殊值>，表示還未進行消費的消息：
- # 窗口 1，消費群組中，taotao 消費者建立阻塞監聽
- XREADGROUP GROUP group1 taotao BLOCK 0 STREAMS person >
- # 窗口 2，消費群組中，yangyang 消費者建立阻塞監聽
- XREADGROUP GROUP group1 yangyang BLOCK 0 STREAMS person >
- # 窗口 3，添加消費消息
- > XADD person 3-1 name tony des 666
- "3-1"
- # 窗口 1，讀取到新消息，此時 窗口 2 沒有任何反應
- > XREADGROUP GROUP group1 taotao BLOCK 0 STREAMS person >
- 1) 1) "person"
- 2) 1) 1) "3-1"
- 2) 1) "name"
- 2) "tony"
- 3) "des"
- 4) "666"
- (77.54s)
- # 窗口 3，再次添加消費消息
- > XADD person 3-2 name james des abc!
- "3-2"
- # 窗口 2，讀取到新消息，此時 窗口 1 沒有任何反應
- > XREADGROUP GROUP group1 yangyang BLOCK 0 STREAMS person >
- 1) 1) "person"
- 2) 1) 1) "3-2"
- 2) 1) "name"
- 2) "james"
- 3) "des"
- 4) "abc!"
- (76.36s)
以上執行流程中，group1 群組中有兩個消費者，當添加兩條消息后，這兩個消費者輪流消費。
消息ACK
消息消費后，為避免再次重復消費，這是需要向服務端發送 ACK，確保消息被消費后的標記。 例如下列情況，我們上面我們將最新兩條消息已進行了消費，但是當我們再次讀取消息時，還是被讀到：
- > XREADGROUP GROUP group1 yangyang STREAMS person 0
- 1) 1) "person"
- 2) 1) 1) "3-2"
- 2) 1) "name"
- 2) "james"
- 3) "des"
- 4) "abc!"
這時，我們使用 XACK 指令告訴服務器，我們已處理的消息:
- XACK key group ID [ID ...]0
讓服務器標記 3-2 已處理：
- > XACK person group1 3-2
- (integer) 1
再次獲取群組讀取消息：
- > XREADGROUP GROUP group1 yangyang STREAMS person 0
- 1) 1) "person"
- 2) (empty list or set)
隊列中沒有了可讀消息。 除了上面以講解到的 API 外，查看消費群組信息可使用 XINFO 指令查看，本文不做分析。
總結
上面對 Streams 常用 API 進行了分析，我們可以感受到 Redis 在消息隊列支持的道路上，也越來越強大。如果使用過它的 PUB/SUB 功能的話，就會感受到 5.x 迭代正是將你的一些痛點進行了優化。
