探究 Redis 4 的 stream 類型

10 月初，Redis 搞了個大新聞。別緊張，是個好消息：Redis 引入了名爲 stream 的新數據類型和對應的命令，大概會在年末正式發佈到 4.x 版本中。像引入新數據類型這樣的變化在 Redis 的發展歷史上很是罕見，因此稱之爲大新聞一點也不爲過。至少不少介紹 Redis 的資料要跟着修訂了。

背景

按做者的介紹，stream 類型的想法深受 Kafka 的 stream 概念的影響，因此瓜熟蒂落沿用了這個名字。固然這並不意味 Redis 將提供 Kafka stream 特性的替代品，它倆依舊是兩種涇渭分明的東西。Redis 的 stream 特性旨在填補 PubSub 和 Blocked list 機制間的空缺，解決這二者不能解決的問題。

Redis 的 PubSub 能夠用來實現簡單的訂閱機制。一個或多個 client 向 Redis 訂閱特定的頻道，當某個 client 向這個頻道發佈消息時，Redis 會把消息發送給訂閱該頻道的 client。須要注意的是，Redis 只負責轉發消息，並不保證訂閱的 client 是否真正收到了消息，好比 client 可能正好掛掉了或者中間出了點網絡問題。在某些狀況下，這種簡單的訂閱機制就夠用了；但在某些狀況下，咱們須要確保消息已經發布出去，PubSub 就不能知足要求。

一個替代的方案是採用 BLPOP 等命令，也即前文提到的 Blocked list。client 調用 BLPOP（或其餘相似的命令），阻塞在特定的頻道上。若是有 client 發佈消息（在這裏，就是 rpush 新的值），被阻塞的 client 就會結束阻塞，獲得新 rpush 進來的值。若是 Redis 無法把新消息發送給 client，那麼這個消息會留在頻道里。當 client 下次從新調用 BLPOP 時，就能拿回這個消息。這個方案聽起來不錯，至少它解決了確保消息發佈的問題。但你可能也想到了，能收到特定頻道的消息的只有一個 client，由於只要某個 client 接收了消息，消息就再也不存在於頻道當中了。而 PubSub 是支持一對多發送消息的。另外一個問題是，每一個 client 只能去獲取最新的消息，對於複雜的操做，BLPOP 等命令便無能爲力了。

stream 就是爲了解決以上問題才提出來的。

用法

遵循其餘數據類型的慣例，操做 stream 類型的鍵的命令都以 X 開頭。
（因爲 stream 特性還沒有正式發佈，且部分特性還處於 TODO 狀態，下面內容確定會有所變動。若是有改動，我會修訂這部分的內容）

添加操做

XADD key [MAXLEN [~] <count>] <ID or *> [field value] [field value] ...

stream 跟 hash 同樣，有 subkey 的概念。上面命令裏的 ID 就是指 subkey。通常狀況下，你不須要指定 ID，僅需提供 * 來讓 Redis 生成一個 ID。Redis 生成的 ID 格式以下：$ms-$seq。其中 $ms 指當前的 13 位毫秒時間戳，$seq 指給定 key 在當前毫秒時間戳下的序列號（從 0 開始），中間以 - 隔開。早前用的分隔符是 .，後來考慮到 xx.yy 這種形式太容易錯看成浮點數了，因此改用 -。若是 Redis 生成 ID 的時候，當前毫秒時間戳跟上一個 ID 的時間戳同樣，它會把序列號加一。假使服務器發生時間回撥的狀況，Redis 會沿用上一個 ID 的時間戳，只是把序列號加一。實際上這種生成 ID 的機制並不是爲了記錄建立的時間，僅僅用於生成遞增的 ID。你也能夠在調用時指定本身生成的 ID。

[field value] ... 這部分指定的是 stream key 對應 ID 的值。每一個 ID 帶的 field 能夠不一樣。

取長度操做

XLEN key

返回長度，就是這樣。

讀取操做

XRANGE key start end [COUNT <n>]

XRANGE 返回某個 stream 給定範圍內的 ID 所對應的值。你能夠經過 COUNT 指定返回的值的最大數目。
舉個例子，像這樣建立兩個 ID：

127.0.0.1:6379> xadd test * apple 1
1507383725597-0
127.0.0.1:6379> xadd test * binana 2
1507383735965-0

以下的 XRANGE 操做可以返回這兩個 ID 的值。

127.0.0.1:6379> xrange test 1507383725597-0 1507383735965-0
1) 1) 1507383725597-0
   2) 1) "apple"
      2) "1"
2) 1) 1507383735965-0
   2) 1) "binana"
      2) "2"

大多數狀況下，你用到的是 - 和 + 這兩個特殊 ID 值，像這樣：xrange test 1507383725597-0 +。前者表示 ID 範圍的起始位置，後者表示 ID 範圍的末尾位置。

XREAD [BLOCK <milliseconds>] [COUNT <count>] [GROUP <groupname> <ttl>]
      [RETRY <milliseconds> <ttl>] STREAMS key_1 key_2 ... key_N
      ID_1 ID_2 ... ID_N

若是想同時讀取多個 stream 的值，須要用到 XREAD。XREAD 可以返回給定多個 stream 的某個 起始ID 以後的數據。我加粗了以後兩個字，由於跟 XRANGE 不一樣，XREAD 不返回 起始ID 的值。你能夠經過 COUNT 指定各個 stream 返回的值的最大數目。

XREAD 的阻塞是可選的，你能夠經過 BLOCK 參數去指定容許阻塞的時間。若是不指定，表示不阻塞，馬上返回 nil。注意這一點跟 BLPOP 不一樣，BLPOP 一類的命令，默認是永久阻塞的。

XREAD 主要的參數是 STREAMS 後面的 key 和 起始ID 列表。key 和 起始ID 須要是一一對應的，有多少個 key 就要指定多少個 起始ID。跟 XRANGE 同樣，起始ID 也能夠是 - 和 + 這樣的特殊值。注意因爲 - 表示 ID 範圍的起始位置，而不是第一個 ID，因此用 - 能夠獲取第一個 ID 的值。除此以外，起始ID 還能夠是 $，表示獲取命令執行以後的新增 ID 的值。顯然，$ 只有跟 BLOCK 一塊兒用纔有意義。

RETRY/GROUP：還沒有實現 TODO。

刪除操做

stream 不支持「改」操做，因此「增刪查改」還剩個「刪」沒講。stream 沒有專門的刪命令。還記得介紹 XADD 時展現的 MAXLEN 參數嗎？在 XADD 命令添加了新的 ID 以後，若是命令指定的 MAXLEN 超過了當前 stream 包含的 ID 的個數，Redis 會刪除多出來的部分。

從新貼下 MAXLEN 的格式：XADD MAXLEN [~] <count> ...。count 決定了 MAXLEN 的值。若是 MAXLEN 和 count 之間沒有插入 ~，表示精確地保留 count 個 ID；若是插入了 ~，表示保留大約 count 個 ID。我會在「實現」這一節解釋所謂的「精確」和「大約」的區別。

用途

stream 很大程度上相似於 Blocked list，可是它的操做更加自由，再也不受限於只能讀取最新的值，也再也不拘束於只能讓單個 client 讀取值。跟 PubSub 相比，stream 容許 client 從新獲取發佈過的值，提供了更強的保障。

實現

Redis 把每一個 stream 實現成以 ID 的值爲 key 的前綴樹，外加 length（當前的 ID 數）等元數據。考慮到默認生成的 ID 是毫秒時間戳+序列號，採用前綴樹的形式能夠節省下大量的空間。畢竟差幾千毫秒的兩個 ID，也會有前九位是徹底相同的。另外前綴樹還容許隨機訪問某個起始ID。

不過並不是每一個 ID 都是獨佔一個節點。每當插入一個新的 ID 時，Redis 會先訪問前綴樹的最大的節點（畢竟 ID 是遞增的），若是這個節點不大於 STREAM_BYTES_PER_LISTPACK（2048字節），新的 ID 會被插入到這個節點裏面；不然纔會建立新的節點。在查找一個 ID 時，Redis 會查找最後一個比該 ID 小的節點，而後從該節點日後遍歷，直到找到該 ID 爲止。在我看來，一個節點裏包含多個 ID 的設計，有利於 ID 遍歷的操做。這種設計避免了在遍歷時頻繁訪問新的節點，更好地利用了 CPU 的本地緩存。

每一個節點具備這樣的結構：

+--------------+---------+---------+--/--+---------+
| master_entry | entry_1 | entry_2 | ... | entry_N |
+--------------+---------+---------+--/--+---------+

其中
master_entry:
+-------+---------+------------+---------+--/--+---------+---------+
| count | deleted | num-fields | field_1 | field_2 | ... | field_N |
+-------+---------+------------+---------+--/--+---------+---------+

entry_x（SAMEFIELDS）:
+-----+--------+-------+-/-+-------+
|flags|entry-id|value-1|...|
+-----+--------+-------+-/-+-------+
或者
+-----+--------+----------+-------+-------+-/-+---+
|flags|entry-id|num-fields|field_1|value_1|...|
+-----+--------+----------+-------+-------+-/-+---+

當節點被建立時，會以第一個插入的 ID 初始化 master_entry 的值。顯然，count 的初始值是 1，deleted 的初始值是 0，num-fields 等於該 ID 對應的 field 數目，後面的多個 field 則是該 ID 對應的 field 數。在插入 master_entry 以後，還會新增一個 entry 來記錄額外的 field 和每一個 field 對應的 value。這個新增的 entry 的 entry-id 取 ID 跟前綴樹節點的 key 的差。第一個 ID 的 entry-id 爲 0，由於當前節點的 key 就是這個 ID，二者不存在差別。以後每插入一個新的 ID，都會更新 master_entry 的 count 數，並插入對應的 entry。固然插入新 ID 的同時也不忘更新 length 等元數據。

前面提到，每一個 ID 帶的 field 能夠不一樣。可是在實際的使用中，每一個 ID 帶的 field 基本是相同的。因此 Redis 作了個優化：若是新增的 ID 的 field 跟 master_entry 徹底同樣，entry 裏面會設置一個名爲 SAMEFIELDS 的 flags，並僅記錄 value 的值。除非新增 ID 的 field 跟 master_entry 有些不一樣，entry 裏面纔會記錄新增 ID 的全部 field 和對應的 value。

最後說一下刪除操做。因爲

1. stream 的刪除操做，只支持保留特定數目的 ID 數
2. stream 會記錄所有的 ID 數（length）
3. stream 的數據結構大致上是一個前綴樹，前綴樹的每一個節點包含 count 個 ID

因此刪除操做，就是

1. 從前日後遍歷，減去每一個前綴樹節點的 count，直到 length 等於 XADD 指定的 MAXLEN，或者減去下一個節點後剩下的 length 會小於 MAXLEN
2. 若是減去某個節點後，剩下的 length 小於 MAXLEN，Redis 會遍歷該節點，設置若干個 entry 的 flags 爲 DELETED 直到 length 等於 MAXLEN，更新 count 和 deleted 兩個域。

若是 XADD 命令指定的 MAXLEN 包含 ~，則表示大約保留 MAXLEN 個 ID。在這種狀況下，Redis 只會完成上面的第一步。換句話說，選擇「大約」能省下對某個節點進行遍歷的時間。