前言
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 迭代正是將你的一些痛點進行了優化。
個人博客： https://ytao.top
關注公眾號 【ytao】，更多原創好文
