Redis精通系列——Stream

本文已收錄于專欄

《Redis精通系列》

上千人點贊收藏，全套Redis學習資料，大廠必備技能！

目錄

1、簡介

2、Stream內部探索

2.1 Stream 結構

2.2 四個唯一

2.3 訊息ID

2.4 訊息內容

3、Stream指令

3.1 指令匯總

3.2 XADD

3.2 XTRIM

3.3 XDEL

3.4 XLEN

3.5 XRANGE

3.6 XREVRANGE

3.7 XREAD

3.8 XGROUP CREATE

3.9 XREADGROUP GROUP

3.10 XACK

3.11 XPENDING

3.11 XCLAIM

3.13 XINFO

4、關于Stream優化記憶體的事情

1、簡介

Stream彌補了Redis作為MQ（message queue）技術選型上的不足之處；Redis 5.0發布的Stream相比Pub/Sub模塊，Stream支持訊息持久化，結合sentinel或cluster使其成為了一個比較可靠的訊息佇列，盡管我認為它很難成為公司MQ的技術選型產品，但是關于Stream的使用和特性（消費組），仍值得一探究竟，
?

Stream對標訊息佇列，因此幾乎具備了MQ所有的特性，以下列出Stream所具有的部分特性：

• 訊息順序存盤
• 訊息ID序列化規則生成
• 訊息的遍歷
• 訊息阻塞/非阻塞式獲取
• 客戶端分組消費訊息
• 訊息確認機制
• 訊息例外機制
• 訊息佇列監控

在文中也會說到Stream的這些特性，
?

2、Stream內部探索

2.1 Stream 結構

在探索Stream的內部結構之前，先看一張清晰的Stream結構圖：

如下是關于上圖的名詞決議：

• Message Content：訊息內容
• Consumer group：消費組，通過XGROUP CREATE 命令創建，一個消費組可以有多個消費者
• Last_delivered_id：游標，每個消費組有一個游標，任意消費者讀取訊息后，游標都會向前移動
• Consumer：消費者，消費組中的消費者
• Pending_ids：狀態變數，每個消費者會有一個狀態變數，用于記錄被當前消費者讀取，但是并未ack的訊息id

2.2 四個唯一

Stream內部維護了一個訊息鏈表，以此使得訊息能夠具有佇列的特性，在Stream中有四個唯一需要了解：

1. 每個Stream都具有唯一的名稱
2. 每個訊息（Message）都具有一個由系統分配或者客戶端指定唯一ID
3. 每個Stream中的消費組（Consumer_Group）具有唯一名稱
4. 每個消費組（Consumer_Group）中的消費者（Consumer）具有唯一名稱

2.3 訊息ID

Stream的訊息ID可以由服務端自動生成，也可以由客戶端傳入，如下圖是自動生成的結構：

系統自動生成的規則

<millisecondsTime>-<sequenceNumber>

millisecondsTime指的是Redis節點服務器的本地時間，如果存在當前的毫秒時間戳比以前已經存在的資料的時間戳小的話（本地時間鐘后跳），那么系統將會采用以前相同的毫秒創建新的ID，
sequenceNumber指的是序列號，在相同的millisecondsTime毫秒下，序列號從0開始遞增，序列號是64位長度，理論上在統一毫秒內生成的資料量無法到達這個級別，因此不用擔心sequenceNumber會不夠用，
?

客戶端顯示傳入規則
Redis對于ID有強制要求，格式必須是-，最小ID為0-1，并且后續ID不能小于前一個ID

2.4 訊息內容

Stream的訊息內容，也就是圖中的Message Content它的結構類似Hash結構，以key-value的形式存在，
?

3、Stream指令

3.1 指令匯總

Stream的指令根據可以分為兩類，分別是訊息佇列相關指令，消費組相關指令，
訊息佇列相關指令：

消費組相關指令：

3.2 XADD

XADD 用于向Stream 佇列中添加訊息，如果指定的Stream 佇列不存在，則該命令執行時會新建一個Stream 佇列，
?

XADD的指令語法：

XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value …]

如下通過XADD展示了定義ID的兩種方式，具體可以看2.3，

3.2 XTRIM

XTRIM 用于對Stream的長度進行限定，
?

XTRIM 的指令語法：

XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]

• MAXLEN 允許的最大長度，如果長度超出則會拋棄佇列前面的訊息
• MINID 允許的最小id，從某個id值開始保留，其余的將會被拋棄

3.3 XDEL

XDEL 用于洗掉訊息，
?

XDEL 的指令語法：

XDEL key ID [ID …]

3.4 XLEN

XLEN 用于獲取Stream 佇列的訊息的長度，
?

XLEN 的指令語法：

XLEN key

3.5 XRANGE

XRANGE 用于獲取訊息串列（可以指定范圍），忽略洗掉的訊息，
?

XRANGE 的指令語法：

XRANGE key start end [COUNT count]

• start 表示開始值，-代表最小值
• end 表示結束值，+代表最大值
• count 表示最多獲取多少個值

3.6 XREVRANGE

XREVRANGE 用于獲取訊息串列（可以指定范圍），忽略洗掉的訊息，與XRANGE 的區別在于，獲取訊息串列元素的方向是相反的，end在前，start在后，
?

XREVRANGE 的指令語法：

XREVRANGE key end start [COUNT count]

3.7 XREAD

XREAD 用于獲取訊息（阻塞/非阻塞），只會回傳大于指定ID的訊息，
?

XREAD 的指令語法：

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key …] ID [ID …]

• COUNT 最多讀取多少條訊息
• BLOCK 是否已阻塞的方式讀取訊息，默認不阻塞，如果milliseconds設定為0，表示永遠阻塞

$代表特殊ID，表示以當前Stream已經存盤的最大的ID作為最后一個ID，當前Stream中不存在大于當前最大ID的訊息，因此此時回傳nil，

0-0代表從最小的ID開始獲取Stream中的訊息，當不指定count，將會回傳Stream中的所有訊息，注意也可以使用0（00/000也都是可以的……），

阻塞方式獲取Stream中的指令，這里演示阻塞獲取一條訊息

?

3.8 XGROUP CREATE

XGROUP CREATE 用于創建消費者組，

XGROUP CREATE 的指令語法：

XGROUP [CREATE key groupname ID|$ [MKSTREAM]] [SETID key groupname ID|$] [DESTROY key groupname] [CREATECONSUMER key groupname consumername] [DELCONSUMER key groupname consumername]

XGROUP CREATE中的指令沒什么復雜的，第一個中括號中的幾個引數最為重要，如下圖兩種方式：

• $表示從Stream尾部開始消費，會忽略Stream中目前已有的資料
• 0表示從Stream頭部開始消費

如果Stream不存在，XGROUP CREATE 語法將會報錯，因此可以得出不允許在不存在的Stream上創建消費者組

3.9 XREADGROUP GROUP

XREADGROUP GROUP 用于讀取消費者組中的訊息，
?

XREADGROUP GROUP 的指令語法：

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key …] ID [ID …]

注意，這里有一個比較重要的知識點，剛開始的時候可能容易搞錯：
>這個特殊符號表示訊息到目前為止，從未傳遞給其他消費者的訊息
0表示指定訊息ID，因為ID均大于0-0（0代指0-0），因此代表從Stream 的佇列頭部開始獲取訊息
?

在如下截圖中，為何第一次 mystream 0 獲取訊息回傳empty，在執行完 mystream > 之后，第二 mystream 0 卻成功的獲取到了訊息，但是很明顯mystream中剛添加了兩條訊息，第一次不應該失敗才對呀？
這是因為，當指定ID進行訊息獲取時，命令將會讓我們訪問我們的歷史待處理訊息（曾被獲取，但是未ack），即傳遞給這個指定消費者（由提供的名稱標識）的訊息集，并且到目前為止從未使用XACK進行確認，

XREADGROUP GROUP 也可以像XREAD 一樣使用阻塞的方式獲取訊息

當向mystream中添加訊息后，阻塞讀回傳

3.10 XACK

XACK 用于標記為“已處理”，

XACK 的指令語法：

XACK key group ID [ID …]

結合 XREADGROUP GROUP 中指定ID的方式只能獲取未ack的未處理訊息的特性，測驗XACK指令，從如下的測驗示例中可以得出兩個結論：

• 訊息首次ack成功，回傳1，ack失敗回傳0
• 3.9中的結論是正確的

3.11 XPENDING

XPENDING 用于列印待處理訊息的詳細資訊，

XPENDING 指令是非常有用的，因為它可以列印待處理訊息的資訊，如果在一個消費者組中存在多個消費者，如果存在部分消費者永久的故障，無法再處理訊息了，我們就可以通過XPENDING 指令來查看指定消費者組中的消費者未ack的訊息，然后轉移給其他消費者進行處理，

XPENDING 的指令語法：

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

XPENDING 回傳值決議：

1. 第一個引數表示當前消費者中待處理訊息的總數
2. 第二個引數表示待處理訊息的最小ID
3. 第三個引數表示待處理訊息的最大ID
4. 第四個引數表示消費者串列和未處理的訊息數量

image.png

?

3.11 XCLAIM

XCLAIM 用于轉移訊息的歸屬權，
?

XCLAIM 的指令語法：

XCLAIM key group consumer min-idle-time ID [ID …] [IDLE ms] [TIME ms-unix-time] [RET

指令引數決議：

• key 表示Stream的名稱
• group 表示需要轉移訊息的歸屬權的消費者組名稱
• consumer 表示接收訊息的消費者名稱
• min-idle-time 表示最小空閑時間，只有后續指定ID的訊息空閑時間大于指定的空閑時間，訊息歸屬權轉移指令才會生效
• ID [] 需要轉移歸屬權的訊息ID，陣列，可以是多個

示例中，將consumer-1中ID為1631719560149-0的未處理的訊息的歸屬權轉移到consumer-2下：

3.13 XINFO

XINFO 用于列印Stream\Consumer\Group的詳細資訊，
?

XINFO 的指令語法：

XINFO [CONSUMERS key groupname] [GROUPS key] [STREAM key] [HELP]

示例列印指定STREAM的詳細訊息

4、關于Stream優化記憶體的事情

使用Stream有兩個點需要注意，如果使用不當都會導致記憶體消耗增大，

1. 待處理訊息過多，訊息未及時ack
2. Stream訊息持續持久化，使用XDEL洗掉訊息

關于第一點，待處理訊息過多，訊息未及時ack，其導致記憶體增加的原因是，Stream會為每個消費者維護一個PEL串列，PEL串列用于存盤處理完但未及時ack的訊息ID，我們在實際使用程序中，處理完的訊息一定要及時ack，也有定時檢查是否有消費者不可用導致訊息堆積的情況，
XPENDING能查詢出消費者中待處理的訊息，就是因為有PEL的存在，

關于第二點，使用XDEL洗掉Stream中不在需要的訊息，其導致記憶體增加的原因是，Stream的XDEL洗掉訊息的指令，并不會從記憶體上洗掉消息，它只是給訊息打上標記位，下次通過XRANGE指令忽略這些訊息而已，因此我們可以設定Stream的最大長度，來解決這個問題，在XADD中使用MAXLEN指定Stream佇列的長度，當訊息超出長度就會將佇列頭訊息清除掉，（不過這種處理方式一定要做到及時處理訊息，避免訊息的丟失，）

XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value …]
?