Stream是Redis 5.0版本引入的一個新的資料類型，它以更抽象的方式模拟日志資料結構，但日志仍然是完整的：就像一個日志檔案，通常實作為以隻附加模式打開的檔案，Redis流主要是一個僅附加資料結構。至少從概念上來講，因為Redis流是一種在記憶體表示的抽象資料類型，他們實作了更加強大的操作，以此來克服日志檔案本身的限制。
Stream是Redis的資料類型中最複雜的，盡管資料類型本身非常簡單，它實作了額外的非強制性的特性：提供了一組允許消費者以阻塞的方式等待生産者向Stream中發送的新消息，此外還有一個名為消費者組的概念。
消費者組最早是由名為Kafka（TM）的流行消息系統引入的。Redis用完全不同的術語重新實作了一個相似的概念，但目标是相同的：允許一組用戶端互相配合來消費同一個Stream的不同部分的消息。
Streams 基礎知識
為了了解Redis Stream是什麼以及如何使用他們，我們将忽略所有的進階特性，從用于操縱和通路它的指令方面來專注于資料結構本身。這基本上是大多數其他Redis資料類型共有的部分，比如Lists，Sets，Sorted Sets等等。然而，需要注意的是Lists還有一個可選的更加複雜的阻塞API，由BLPOP等相似的指令導出。是以從這方面來說，Streams跟Lists并沒有太大的不同，隻是附加的API更複雜、更強大。
因為Streams是隻附加資料結構，基本的寫指令，叫XADD，向指定的Stream追加一個新的條目。一個Stream條目不是簡單的字元串，而是由一個或多個鍵值對組成的。這樣一來，Stream的每一個條目就已經是結構化的，就像以CSV格式寫的隻附加檔案一樣，每一行由多個逗号割開的字段組成。
> XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0
上面的例子中，調用了XADD指令往名為
mystream
的Stream中添加了一個條目
sensor-id: 123, temperature: 19.8
，使用了自動生成的條目ID，也就是指令傳回的值，具體在這裡是
1518951480106-0
。指令的第一個參數是key的名稱
mystream
，第二個參數是用于唯一确認Stream中每個條目的條目ID。然而，在這個例子中，我們傳入的參數值是
*
，因為我們希望由Redis伺服器為我們自動生成一個新的ID。每一個新的ID都會單調增長，簡單來講就是，每次新添加的條目都會擁有一個比其它所有條目更大的ID。由伺服器自動生成ID幾乎總是我們所想要的，需要顯式指定ID的情況非常少見。我們稍後會更深入地讨論這個問題。實際上每個Stream條目擁有一個ID與日志檔案具有另一種相似性，即使用行号或者檔案中的位元組偏移量來識别一個給定的條目。回到我們的XADD例子中，跟在key和ID後面的參數是組成我們的Stream條目的鍵值對。
使用XLEN指令來擷取一個Stream的條目數量：
> XLEN mystream
(integer) 1
條目 ID
條目ID由XADD指令傳回，并且可以唯一的辨別給定Stream中的每一個條目，由兩部分組成：
<millisecondsTime>-<sequenceNumber>
毫秒時間部分實際是生成Stream ID的Redis節點的伺服器本地時間，但是如果目前毫秒時間戳比以前的條目時間戳小的話，那麼會使用以前的條目時間，是以即便是伺服器時鐘向後跳，單調增長ID的特性仍然會保持不變。序列号用于以相同毫秒建立的條目。由于序列号是64位的，是以實際上對于在同一毫秒内生成的條目數量是沒有限制的。
這樣的ID格式也許最初看起來有點奇怪，也許溫柔的讀者會好奇為什麼時間會是ID的一部分。其實是因為Redis Streams支援按ID進行範圍查詢。由于ID與生成條目的時間相關，是以可以很容易地按時間範圍進行查詢。我們在後面講到XRANGE指令時，很快就能明白這一點。
如果由于某些原因，使用者需要與時間無關但實際上與另一個外部系統ID關聯的增量ID，就像前面所說的，XADD指令可以帶上一個顯式的ID，而不是使用通配符
*
來自動生成，如下所示：
> XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2
請注意，在這種情況下，最小ID為0-1，并且指令不接受等于或小于前一個ID的ID：
> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item
從Streams中擷取資料
現在我們終于能夠通過XADD指令向我們的Stream中追加條目了。然而，雖然往Stream中追加資料非常明顯，但是為了提取資料而查詢Stream的方式并不是那麼明顯，如果我們繼續使用日志檔案進行類比，一種顯而易見的方式是模拟我們通常使用Unix指令
tail -f
來做的事情，也就是，我們可以開始監聽以擷取追加到Stream的新消息。需要注意的是，不像Redis的阻塞清單，一個給定的元素隻能到達某一個使用了冒泡風格的阻塞用戶端，比如使用類似BLPOP的指令，在Streams中我們希望看到的是多個消費者都能看到追加到Stream中的新消息，就像許多的
tail -f
程序能同時看到追加到日志檔案的内容一樣。用傳統術語來講就是我們希望Streams可以扇形分發消息到多個用戶端。
然而，這隻是其中一種可能的通路模式。我們還可以使用一種完全不同的方式來看待一個Stream：不是作為一個消息傳遞系統，而是作為一個時間序列存儲。在這種情況下，也許使附加新消息也非常有用，但是另一種自然查詢模式是通過時間範圍來擷取消息，或者使用一個遊标來增量周遊所有的曆史消息。這絕對是另一種有用的通路模式。
最後，如果我們從消費者的角度來觀察一個Stream，我們也許想要以另外一種方式來通路它，那就是，作為一個可以分區到多個處理此類消息的多個消費者的消息流，以便消費者組隻能看到到達單個流的消息的子集。
Redis Streams通過不同的指令支援所有上面提到的三種通路模式。接下來的部分将展示所有這些模式，從最簡單和更直接的使用：範圍查詢開始。
按範圍查詢: XRANGE 和 XREVRANGE
要根據範圍查詢Stream，我們隻需要提供兩個ID，即start 和 end。傳回的區間資料将會包括ID是start和end的元素，是以區間是完全包含的。兩個特殊的ID
-
和
+
分别表示可能的最小ID和最大ID。
> XRANGE mystream - +
1) 1) 1518951480106-0
2) 1) "sensor-id"
2) "1234"
3) "temperature"
4) "19.8"
2) 1) 1518951482479-0
2) 1) "sensor-id"
2) "9999"
3) "temperature"
4) "18.2"
傳回的每個條目都是有兩個元素的數組：ID和鍵值對清單。我們已經說過條目ID與時間有關系，因為在字元
-
左邊的部分是建立Stream條目的本地節點上的Unix毫秒時間，即條目建立的那一刻（請注意：Streams的複制使用的是完全詳盡的XADD指令，是以從節點将具有與主節點相同的ID）。這意味着我可以使用XRANGE查詢一個時間範圍。然而為了做到這一點，我可能想要省略ID的序列号部分：如果省略，區間範圍的開始序列号将預設為0，結束部分的序列号預設是有效的最大序列号。這樣一來，僅使用兩個Unix毫秒時間去查詢，我們就可以得到在那段時間内産生的所有條目（包含開始和結束）。例如，我可能想要查詢兩毫秒時間，可以這樣使用：
> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
2) 1) "sensor-id"
2) "1234"
3) "temperature"
4) "19.8"
我在這個範圍内隻有一個條目，然而在實際資料集中，我可以查詢數小時的範圍，或者兩毫秒之間包含了許多的項目，傳回的結果集很大。是以，XRANGE指令支援在最後放一個可選的COUNT選項。通過指定一個count，我可以隻擷取前面N個項目。如果我想要更多，我可以拿傳回的最後一個ID，在序列号部分加1，然後再次查詢。我們在下面的例子中看到這一點。我們開始使用XADD添加10個項目（我這裡不具體展示，假設流
mystream
已經填充了10個項目）。要開始我的疊代，每個指令隻擷取2個項目，我從全範圍開始，但count是2。
> XRANGE mystream - + COUNT 2
1) 1) 1519073278252-0
2) 1) "foo"
2) "value_1"
2) 1) 1519073279157-0
2) 1) "foo"
2) "value_2"
為了繼續下兩個項目的疊代，我必須選擇傳回的最後一個ID，即
1519073279157-0
，并且在ID序列号部分加1。請注意，序列号是64位的，是以無需檢查溢出。在這個例子中，我們得到的結果ID是
1519073279157-1
，現在可以用作下一次XRANGE調用的新的start參數：
> XRANGE mystream 1519073279157-1 + COUNT 2
1) 1) 1519073280281-0
2) 1) "foo"
2) "value_3"
2) 1) 1519073281432-0
2) 1) "foo"
2) "value_4"
依此類推。由于XRANGE的查找複雜度是O(log(N))，是以O(M)傳回M個元素，這個指令在count較小時，具有對數時間複雜度，這意味着每一步疊代速度都很快。是以XRANGE也是事實上的流疊代器并且不需要XSCAN指令。
XREVRANGE指令與XRANGE相同，但是以相反的順序傳回元素，是以XREVRANGE的實際用途是檢查一個Stream中的最後一項是什麼：
> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
2) 1) "foo"
2) "value_10"
請注意：XREVRANGE指令以相反的順序擷取start 和 stop參數。
使用XREAD監聽新項目
當我們不想按照Stream中的某個範圍通路項目時，我們通常想要的是訂閱到達Stream的新項目。這個概念可能與Redis中你訂閱頻道的Pub/Sub或者Redis的阻塞清單有關，在這裡等待某一個key去擷取新的元素，但是這跟你消費Stream有着根本的不同：
- 一個Stream可以擁有多個用戶端（消費者）在等待資料。預設情況下，對于每一個新項目，都會被分發到等待給定Stream的資料的每一個消費者。這個行為與阻塞清單不同，每個消費者都會擷取到不同的元素。但是，扇形分發到多個消費者的能力與Pub/Sub相似。
- 雖然在Pub/Sub中的消息是fire and forget并且從不存儲，以及使用阻塞清單時，當一個用戶端收到消息時，它會從清單中彈出（有效删除），Stream從跟本上以一種不同的方式工作。所有的消息都被無限期地附加到Stream中（除非使用者明确地要求删除這些條目）：不同的消費者通過記住收到的最後一條消息的ID，從其角度知道什麼是新消息。
- Streams 消費者組提供了一種Pub/Sub或者阻塞清單都不能實作的控制級别，同一個Stream不同的群組，顯式地确認已經處理的項目，檢查待處理的項目的能力，申明未處理的消息，以及每個消費者擁有連貫曆史可見性，單個用戶端隻能檢視自己過去的消息曆史記錄。
提供監聽到達Stream的新消息的能力的指令稱為XREAD。比XRANGE要更複雜一點，是以我們将從簡單的形式開始，稍後将提供整個指令布局。
> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
2) 1) 1) 1519073278252-0
2) 1) "foo"
2) "value_1"
2) 1) 1519073279157-0
2) 1) "foo"
2) "value_2"
以上是XREAD的非阻塞形式。注意COUNT選項并不是必需的，實際上這個指令唯一強制的選項是STREAMS，指定了一組key以及調用者已經看到的每個Stream相應的最大ID，以便該指令僅向用戶端提供ID大于我們指定ID的消息。
在上面的指令中，我們寫了
STREAMS mystream 0
，是以我們想要流
mystream
中所有ID大于
0-0
的消息。正如你在上面的例子中所看到的，指令傳回了鍵名，因為實際上可以通過傳入多個key來同時從不同的Stream中讀取資料。我可以寫一下，例如：
STREAMS mystream otherstream 0 0
。注意在STREAMS選項後面，我們需要提供鍵名稱，以及之後的ID。是以，STREAMS選項必須始終是最後一個。
除了XREAD可以同時通路多個Stream這一事實，以及我們能夠指定我們擁有的最後一個ID來擷取之後的新消息，在個簡單的形式中，這個指令并沒有做什麼跟XRANGE有太大差別的事情。然而，有趣的部分是我們可以通過指定BLOCK參數，輕松地将XREAD 變成一個 阻塞指令：
> XREAD BLOCK 0 STREAMS mystream $
請注意，在上面的例子中，除了移除COUNT以外，我指定了新的BLOCK選項，逾時時間為0毫秒（意味着永不逾時）。此外，我并沒有給流
mystream
傳入一個正常的ID，而是傳入了一個特殊的ID
$
。這個特殊的ID意思是XREAD應該使用流
mystream
已經存儲的最大ID作為最後一個ID。以便我們僅接收從我們開始監聽時間以後的新消息。這在某種程度上相似于Unix指令
tail -f
。
請注意當使用BLOCK選項時，我們不必使用特殊ID
$
。我們可以使用任意有效的ID。如果指令能夠立即處理我們的請求而不會阻塞，它将執行此操作，否則它将阻止。通常如果我們想要從新的條目開始消費Stream，我們以
$
開始，接着繼續使用接收到的最後一條消息的ID來發起下一次請求，依此類推。
XREAD的阻塞形式同樣可以監聽多個Stream，隻需要指定多個鍵名即可。如果請求可以同步提供，因為至少有一個流的元素大于我們指定的相應ID，則傳回結果。否則，該指令将阻塞并将傳回擷取新資料的第一個流的項目（根據提供的ID）。
跟阻塞清單的操作類似，從等待資料的用戶端角度來看，阻塞流讀取是公正的，由于語義是FIFO樣式。阻塞給定Stream的第一個用戶端是第一個在新項目可用時将被解除阻塞的用戶端。
XREAD指令沒有除了COUNT 和 BLOCK以外的其他選項，是以它是一個非常基本的指令，具有特定目的來攻擊消費者一個或多個流。使用消費者組API可以用更強大的功能來消費Stream，但是通過消費者組讀取是通過另外一個不同的指令來實作的，稱為XREADGROUP。本指南的下一節将介紹。
消費者組
當手頭的任務是從不同的用戶端消費同一個Stream，那麼XREAD已經提供了一種方式可以扇形分發到N個用戶端，還可以使用從節點來提供更多的讀取可伸縮性。然而，在某些問題中，我們想要做的不是向許多用戶端提供相同的消息流，而是從同一流向許多用戶端提供不同的消息子集。這很有用的一個明顯的例子是處理消息的速度很慢：能夠讓N個不同的用戶端接收流的不同部分，通過将不同的消息路由到準備做更多工作的不同用戶端來擴充消息處理工作。
實際上，假如我們想象有三個消費者C1，C2，C3，以及一個包含了消息1, 2, 3, 4, 5, 6, 7的Stream，我們想要按如下圖表的方式處理消息：
1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1
為了獲得這個效果，Redis使用了一個名為消費者組的概念。非常重要的一點是，從實作的角度來看，Redis的消費者組與Kafka (TM) 消費者組沒有任何關系，它們隻是從實施的概念上來看比較相似，是以我決定不改變最初普及這種想法的軟體産品已有的術語。
消費者組就像一個僞消費者，從流中擷取資料，實際上為多個消費者提供服務，提供某些保證：
- 每條消息都提供給不同的消費者，是以不可能将相同的消息傳遞給多個消費者。
- 消費者在消費者組中通過名稱來識别，該名稱是實施消費者的客戶必須選擇的區分大小寫的字元串。這意味着即便斷開連接配接過後，消費者組仍然保留了所有的狀态，因為用戶端會重新申請成為相同的消費者。 然而，這也意味着由用戶端提供唯一的辨別符。
- 每一個消費者組都有一個第一個ID永遠不會被消費的概念，這樣一來，當消費者請求新消息時，它能提供以前從未傳遞過的消息。
- 消費消息需要使用特定的指令進行顯式确認，表示：這條消息已經被正确處理了，是以可以從消費者組中逐出。
- 消費者組跟蹤所有目前所有待處理的消息，也就是，消息被傳遞到消費者組的一些消費者，但是還沒有被确認為已處理。由于這個特性，當通路一個Stream的曆史消息的時候，每個消費者将隻能看到傳遞給它的消息。
在某種程度上，消費者組可以被想象為關于Stream的一些狀态：
| consumer_group_name: mygroup |
| consumer_group_stream: somekey |
| last_delivered_id: 1292309234234-92 |
| |
| consumers: |
| "consumer-1" with pending messages |
| 1292309234234-4 |
| 1292309234232-8 |
| "consumer-42" with pending messages |
| ... (and so forth) |
如果你從這個視角來看，很容易了解一個消費者組能做什麼，如何做到向給消費者提供他們的曆史待處理消息，以及當消費者請求新消息的時候，是如何做到隻發送ID大于
last_delivered_id
的消息的。同時，如果你把消費者組看成Redis Stream的輔助資料結構，很明顯單個Stream可以擁有多個消費者組，每個消費者組都有一組消費者。實際上，同一個Stream甚至可以通過XREAD讓用戶端在沒有消費者組的情況下讀取，同時有用戶端通過XREADGROUP在不同的消費者組中讀取。
現在是時候放大來檢視基本的消費者組指令了，具體如下：
- XGROUP 用于建立，摧毀或者管理消費者組。
- XREADGROUP 用于通過消費者組從一個Stream中讀取。
- XACK 是允許消費者将待處理消息标記為已正确處理的指令。
建立一個消費者組
假設我已經存在類型流的
mystream
，為了建立消費者組，我隻需要做：
> XGROUP CREATE mystream mygroup $
OK
請注意：目前還不能為不存在的Stream建立消費者組，但有可能在不久的将來我們會給XGROUP指令增加一個選項，以便在這種場景下可以建立一個空的Stream。
如你所看到的上面這個指令，當建立一個消費者組的時候，我們必須指定一個ID，在這個例子中ID是
$
。這是必要的，因為消費者組在其他狀态中必須知道在第一個消費者連接配接時接下來要服務的消息，即消費者組建立完成時的最後消息ID是什麼？如果我們就像上面例子一樣，提供一個
$
，那麼隻有從現在開始到達Stream的新消息才會被傳遞到消費者組中的消費者。如果我們指定的消息ID是
，那麼消費者組将會開始消費這個Stream中的所有曆史消息。當然，你也可以指定任意其他有效的ID。你所知道的是，消費者組将開始傳遞ID大于你所指定的ID的消息。因為
$
表示Stream中目前最大ID的意思，指定
$
會有隻消費新消息的效果。
現在消費者組建立好了，我們可以使用XREADGROUP指令立即開始嘗試通過消費者組讀取消息。我們會從消費者那裡讀到，假設指定消費者分别是Alice和Bob，來看看系統會怎樣傳回不同消息給Alice和Bob。
XREADGROUP和XREAD非常相似，并且提供了相同的BLOCK選項，除此以外還是一個同步指令。但是有一個強制的選項必須始終指定，那就是GROUP，并且有兩個參數：消費者組的名字，以及嘗試讀取的消費者的名字。選項COUNT仍然是支援的，并且與XREAD指令中的用法相同。
在開始從Stream中讀取之前，讓我們往裡面放一些消息：
> XADD mystream * message apple
1526569495631-0
> XADD mystream * message orange
1526569498055-0
> XADD mystream * message strawberry
1526569506935-0
> XADD mystream * message apricot
1526569535168-0
> XADD mystream * message banana
1526569544280-0
請注意：在這裡消息是字段名稱，水果是關聯的值，記住Stream中的每一項都是小字典。
現在是時候嘗試使用消費者組讀取了：
> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
2) 1) 1) 1526569495631-0
2) 1) "message"
2) "apple"
XREADGROUP的響應内容就像XREAD一樣。但是請注意上面提供的
GROUP <group-name> <consumer-name>
，這表示我想要使用消費者組
mygroup
從Stream中讀取，我是消費者
Alice
。每次消費者使用消費者組中執行操作時，都必須要指定可以這個消費者組中唯一辨別它的名字。
在以上指令行中還有另外一個非常重要的細節，在強制選項STREAMS之後，鍵
mystream
請求的ID是特殊的ID
>
。這個特殊的ID隻在消費者組的上下文中有效，其意思是：消息到目前為止從未傳遞給其他消費者。
這幾乎總是你想要的，但是也可以指定一個真實的ID，比如
或者任何其他有效的ID，在這個例子中，我們請求XREADGROUP隻提供給我們曆史待處理的消息，在這種情況下，将永遠不會在組中看到新消息。是以基本上XREADGROUP可以根據我們提供的ID有以下行為：
如果ID是特殊ID
>
，那麼指令将會傳回到目前為止從未傳遞給其他消費者的新消息，這有一個副作用，就是會更新消費者組的最後ID。 如果ID是任意其他有效的數字ID，那麼指令将會讓我們通路我們的曆史待處理消息。即傳遞給這個指定消費者（由提供的名稱辨別）的消息集，并且到目前為止從未使用XACK進行确認。
我們可以立即測試此行為，指定ID為0，不帶任何COUNT選項：我們隻會看到唯一的待處理消息，即關于apples的消息：
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
2) 1) 1) 1526569495631-0
2) 1) "message"
2) "apple"
但是，如果我們确認這個消息已經處理，它将不再是曆史待處理消息的一部分，是以系統将不再報告任何消息：
> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
2) (empty list or set)
如果你還不清楚XACK是如何工作的，請不用擔心，這個概念隻是已處理的消息不再是我們可以通路的曆史記錄的一部分。
現在輪到Bob來讀取一些東西了：
> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
2) 1) 1) 1526569498055-0
2) 1) "message"
2) "orange"
2) 1) 1526569506935-0
2) 1) "message"
2) "strawberry"
Bob要求最多兩條消息，并通過同一消費者組
mygroup
讀取。是以發生的是Redis僅報告新消息。正如你所看到的，消息”apple”未被傳遞，因為它已經被傳遞給Alice，是以Bob擷取到了orange和strawberry，以此類推。
這樣，Alice，Bob以及這個消費者組中的任何其他消費者，都可以從相同的Stream中讀取到不同的消息，讀取他們尚未處理的曆史消息，或者标記消息為已處理。這允許建立不同的拓撲和語義來從Stream中消費消息。
有幾件事需要記住：
- 消費者是在他們第一次被提及的時候自動建立的，不需要顯式建立。
- 即使使用XREADGROUP，你也可以同時從多個key中讀取，但是要讓其工作，你需要給每一個Stream建立一個名稱相同的消費者組。這并不是一個常見的需求，但是需要說明的是，這個功能在技術上是可以實作的。
- XREADGROUP指令是一個寫指令，因為當它從Stream中讀取消息時，消費者組被修改了，是以這個指令隻能在master節點調用。
使用Ruby語言編寫的使用使用者組的消費者實作示例如下。 Ruby代碼的編寫方式，幾乎對使用任何其他語言程式設計的程式員或者不懂Ruby的人來說，都是清晰可讀的：
require 'redis'
if ARGV.length == 0 puts "Please specify a consumer name" exit 1 end ConsumerName = ARGV[0] GroupName = "mygroup" r = Redis.new def process_message(id,msg) puts "[#{ConsumerName}] #{id} = #{msg.inspect}" end $lastid = '0-0' puts "Consumer #{ConsumerName} starting..." check_backlog = true while true # Pick the ID based on the iteration: the first time we want to # read our pending messages, in case we crashed and are recovering. # Once we consumer our history, we can start getting new messages. if check_backlog myid = $lastid else myid = '>' end items = r.xreadgroup('GROUP',GroupName,ConsumerName,'BLOCK','2000','COUNT','10','STREAMS',:my_stream_key,myid) if items == nil puts "Timeout!" next end # If we receive an empty reply, it means we were consuming our history # and that the history is now empty. Let's start to consume new messages. check_backlog = false if items[0][1].length == 0 items[0][1].each{|i| id,fields = i # Process the message process_message(id,fields) # Acknowledge the message as processed r.xack(:my_stream_key,GroupName,id) $lastid = id } end
正如你所看到的，這裡的想法是開始消費曆史消息，即我們的待處理消息清單。這很有用，因為消費者可能已經崩潰，是以在重新啟動時，我們想要重新讀取那些已經傳遞給我們但還沒有确認的消息。通過這種方式，我們可以多次或者一次處理消息（至少在消費者失敗的場景中是這樣，但是這也受到Redis持久化和複制的限制，請參閱有關此主題的特定部分）。
消耗曆史消息後，我們将得到一個空的消息清單，我們可以切換到
>
，使用特殊ID來消費新消息。
從永久性失敗中恢複
上面的例子允許我們編寫多個消費者參與同一個消費者組，每個消費者擷取消息的一個子集進行處理，并且在故障恢複時重新讀取各自的待處理消息。然而在現實世界中，消費者有可能永久地失敗并且永遠無法恢複。由于任何原因停止後，消費者的待處理消息會發生什麼呢？
Redis的消費者組提供了一個專門針對這種場景的特性，用以認領給定消費者的待處理消息，這樣一來，這些消息就會改變他們的所有者，并且被重新配置設定給其他消費者。這個特性是非常明确的，消費者必須檢查待處理消息清單，并且必須使用特殊指令來認領特定的消息，否則伺服器将把待處理的消息永久配置設定給舊消費者，這樣不同的應用程式就可以選擇是否使用這樣的特性，以及使用它的方式。
這個過程的第一步是使用一個叫做XPENDING的指令，這個指令提供消費者組中待處理條目的可觀察性。這是一個隻讀指令，它總是可以安全地調用，不會改變任何消息的所有者。在最簡單的形式中，調用這個指令隻需要兩個參數，即Stream的名稱和消費者組的名稱。
> XPENDING mystream mygroup
1) (integer) 2
2) 1526569498055-0
3) 1526569506935-0
4) 1) 1) "Bob"
2) "2"
當以這種方式調用的時候，指令隻會輸出給定消費者組的待處理消息總數（在本例中是兩條消息），所有待處理消息中的最小和最大的ID，最後是消費者清單和每個消費者的待處理消息數量。我們隻有Bob有兩條待處理消息，因為Alice請求的唯一一條消息已使用XACK确認了。
我們可以通過給XPENDING指令傳遞更多的參數來擷取更多資訊，完整的指令簽名如下：
XPENDING <key> <groupname> [<start-id> <end-id> <count> [<conusmer-name>]]
通過提供一個開始和結束ID（可以隻是
-
和
+
，就像XRANGE一樣），以及一個控制指令傳回的資訊量的數字，我們可以了解有關待處理消息的更多資訊。如果我們想要将輸出限制為僅針對給定使用者組的待處理消息，可以使用最後一個可選參數，即消費者組的名稱，但我們不會在以下示例中使用此功能。
> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-0
2) "Bob"
3) (integer) 74170458
4) (integer) 1
2) 1) 1526569506935-0
2) "Bob"
3) (integer) 74170458
4) (integer) 1
現在我們有了每一條消息的詳細資訊：消息ID，消費者名稱，空閑時間（機關是毫秒，意思是：自上次将消息傳遞給某個消費者以來經過了多少毫秒），以及每一條給定的消息被傳遞了多少次。我們有來自Bob的兩條消息，它們空閑了74170458毫秒，大概20個小時。
請注意，沒有人阻止我們檢查第一條消息内容是什麼，使用XRANGE即可。
> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
2) 1) "message"
2) "orange"
我們隻需要在參數中重複兩次相同的ID。現在我們有了一些想法，Alice可能會根據過了20個小時仍然沒有處理這些消息，來判斷Bob可能無法及時恢複，是以現在是時候認領這些消息，并繼續代替Bob處理了。為了做到這一點，我們使用XCLAIM指令。
這個指令非常的複雜，并且在其完整形式中有很多選項，因為它用于複制消費者組的更改，但我們隻使用我們通常需要的參數。在這種情況下，它就像調用它一樣簡單：
XCLAIM <key> <group> <consumer> <min-idle-time> <ID-1> <ID-2> ... <ID-N>
基本上我們說，對于這個特定的Stream和消費者組，我希望指定的ID的這些消息可以改變他們的所有者，并将被配置設定到指定的消費者
<consumer>
。但是，我們還提供了最小空閑時間，是以隻有在上述消息的空閑時間大于指定的空閑時間時，操作才會起作用。這很有用，因為有可能兩個用戶端會同時嘗試認領一條消息：
Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
然而認領一條消息的副作用是會重置它的閑置時間！并将增加其傳遞次數的計數器，是以上面第二個用戶端的認領會失敗。通過這種方式，我們可以避免對消息進行簡單的重新處理（即使是在一般情況下，你仍然不能獲得準确的一次處理）。
下面是指令執行的結果：
> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
2) 1) "message"
2) "orange"
Alice成功認領了該消息，現在可以處理并确認消息，盡管原來的消費者還沒有恢複，也能往前推動。
從上面的例子很明顯能看到，作為成功認領了指定消息的副作用，XCLAIM指令也傳回了消息資料本身。但這不是強制性的。可以使用JUSTID選項，以便僅傳回成功認領的消息的ID。如果你想減少用戶端和伺服器之間的帶寬使用量的話，以及考慮指令的性能，這會很有用，并且你不會對消息感興趣，因為稍後你的消費者的實作方式将不時地重新掃描曆史待處理消息。
認領也可以通過一個獨立的程序來實作：這個程序隻負責檢查待處理消息清單，并将空閑的消息配置設定給看似活躍的消費者。可以通過Redis Stream的可觀察特性獲得活躍的消費者。這是下一個章節的主題。
消息認領及傳遞計數器
在XPENDING的輸出中，你所看到的計數器是每一條消息的傳遞次數。這樣的計數器以兩種方式遞增：消息通過XCLAIM成功認領時，或者調用XREADGROUP通路曆史待處理消息時。
當出現故障時，消息被多次傳遞是很正常的，但最終它們通常會得到處理。但有時候處理特定的消息會出現問題，因為消息會以觸發處理代碼中的bug的方式被損壞或修改。在這種情況下，消費者處理這條特殊的消息會一直失敗。因為我們有傳遞嘗試的計數器，是以我們可以使用這個計數器來檢測由于某些原因根本無法處理的消息。是以一旦消息的傳遞計數器達到你給定的值，比較明智的做法是将這些消息放入另外一個Stream，并給系統管理者發送一條通知。這基本上是Redis Stream實作的dead letter概念的方式。
Streams 的可觀察性
缺乏可觀察性的消息系統很難處理。不知道誰在消費消息，哪些消息待處理，不知道給定Stream的活躍消費者組的集合，使得一切都不透明。是以，Redis Stream和消費者組都有不同的方式來觀察正在發生的事情。我們已經介紹了XPENDING，它允許我們檢查在給定時刻正在處理的消息清單，以及它們的空閑時間和傳遞次數。
但是，我們可能希望做更多的事情，XINFO指令是一個可觀察性接口，可以與子指令一起使用，以擷取有關Stream或消費者組的資訊。
這個指令使用子指令來顯示有關Stream和消費者組的狀态的不同資訊，比如使用**XINFO STREAM **可以報告關于Stream本身的資訊。
> XINFO STREAM mystream
1) length
2) (integer) 13
3) radix-tree-keys
4) (integer) 1
5) radix-tree-nodes
6) (integer) 2
7) groups
8) (integer) 2
9) first-entry
10) 1) 1524494395530-0
2) 1) "a"
2) "1"
3) "b"
4) "2"
11) last-entry
12) 1) 1526569544280-0
2) 1) "message"
2) "banana"
輸出顯示了有關如何在内部編碼Stream的資訊，以及顯示了Stream的第一條和最後一條消息。另一個可用的資訊是與這個Stream相關聯的消費者組的數量。我們可以進一步挖掘有關消費者組的更多資訊。
> XINFO GROUPS mystream
1) 1) name
2) "mygroup"
3) consumers
4) (integer) 2
5) pending
6) (integer) 2
2) 1) name
2) "some-other-group"
3) consumers
4) (integer) 1
5) pending
6) (integer) 0
正如你在這裡和前面的輸出中看到的，XINFO指令輸出一系列鍵值對。因為這是一個可觀察性指令，允許人類使用者立即了解報告的資訊，并允許指令通過添加更多字段來報告更多資訊，而不會破壞與舊用戶端的相容性。其他更高帶寬效率的指令，比如XPENDING，隻報告沒有字段名稱的資訊。
上面例子中的輸出（使用了子指令GROUPS）應該能清楚地觀察字段名稱。我們可以通過檢查在此類消費者組中注冊的消費者，來更詳細地檢查特定消費者組的狀态。
> XINFO CONSUMERS mystream mygroup
1) 1) name
2) "Alice"
3) pending
4) (integer) 1
5) idle
6) (integer) 9104628
2) 1) name
2) "Bob"
3) pending
4) (integer) 1
5) idle
6) (integer) 83841983
如果你不記得指令的文法，隻需要檢視指令本身的幫助：
> XINFO HELP
1) XINFO <subcommand> arg arg ... arg. Subcommands are:
2) CONSUMERS <key> <groupname> -- Show consumer groups of group <groupname>.
3) GROUPS <key> -- Show the stream consumer groups.
4) STREAM <key> -- Show information about the stream.
5) HELP -- Print this help.
與Kafka（TM）分區的差異
Redis Stream的消費者組可能類似于基于Kafka（TM）分區的消費者組，但是要注意Redis Stream實際上非常不同。分區僅僅是邏輯的，并且消息隻是放在一個Redis鍵中，是以不同用戶端的服務方式取決于誰準備處理新消息，而不是從哪個分區用戶端讀取。例如，如果消費者C3在某一點永久故障，Redis會繼續服務C1和C2，将新消息送達，就像現在隻有兩個邏輯分區一樣。
類似地，如果一個給定的消費者在處理消息方面比其他消費者快很多，那麼這個消費者在相同機關時間内按比例會接收更多的消息。這是有可能的，因為Redis顯式地追蹤所有未确認的消息，并且記住了誰接收了哪些消息，以及第一條消息的ID從未傳遞給任何消費者。
但是，這也意味着在Redis中，如果你真的想把同一個Stream的消息分區到不同的Redis執行個體中，你必須使用多個key和一些分區系統，比如Redis叢集或者特定應用程式的分區系統。單個Redis Stream不會自動分區到多個執行個體上。
我們可以說，以下是正确的：
- 如果你使用一個Stream對應一個消費者，則消息是按順序處理的。
- 如果你使用N個Stream對應N個消費者，那麼隻有給定的消費者hits N個Stream的子集，你可以擴充上面的模型來實作。
- 如果你使用一個Stream對應多個消費者，則對N個消費者進行負載平衡，但是在那種情況下，有關同一邏輯項的消息可能會無序消耗，因為給定的消費者處理消息3可能比另一個消費者處理消息4要快。
是以基本上Kafka分區更像是使用了N個不同的Redis鍵。而Redis消費者組是一個将給定Stream的消息負載均衡到N個不同消費者的服務端負載均衡系統。
設定Streams的上限
許多應用并不希望将資料永久收集到一個Stream。有時在Stream中指定一個最大項目數很有用，之後一旦達到給定的大小，将資料從Redis中移到不那麼快的非記憶體存儲是有用的，适合用來記錄未來幾十年的曆史資料。Redis Stream對此有一定的支援。這就是XADD指令的MAXLEN選項，這個選項用起來很簡單：
> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
2) 1) "value"
2) "2"
2) 1) 1526655000369-0
2) 1) "value"
2) "3"
如果使用MAXLEN選項，當Stream的達到指定長度後，老的條目會自動被驅逐，是以Stream的大小是恒定的。目前還沒有選項讓Stream隻保留給定數量的條目，因為為了一緻地運作，這樣的指令必須為了驅逐條目而潛在地阻塞很長時間。比如可以想象一下如果存在插入尖峰，然後是長暫停，以及另一次插入，全都具有相同的最大時間。Stream會阻塞來驅逐在暫停期間變得太舊的資料。是以，使用者需要進行一些規劃并了解Stream所需的最大長度。此外，雖然Stream的長度與記憶體使用是成正比的，但是按時間來縮減不太容易控制和預測：這取決于插入速率，該變量通常随時間變化（當它不變化時，那麼按尺寸縮減是微不足道的）。
然而使用MAXLEN進行修整可能很昂貴：Stream由宏節點表示為基數樹，以便非常節省記憶體。改變由幾十個元素組成的單個宏節點不是最佳的。是以可以使用以下特殊形式提供指令：
XADD mystream MAXLEN ~ 1000 * ... entry fields here ...
在選項MAXLEN和實際計數中間的參數
~
的意思是，我不是真的需要精确的1000個項目。它可以是1000或者1010或者1030，隻要保證至少儲存1000個項目就行。通過使用這個參數，僅當我們移除整個節點的時候才執行修整。這使得指令更高效，而且這也是我們通常想要的。
還有XTRIM指令可用，它做的事情與上面講到的MAXLEN選項非常相似，但是這個指令不需要添加任何其他參數，可以以獨立的方式與Stream一起使用。
> XTRIM mystream MAXLEN 10
或者，對于XADD選項：
> XTRIM mystream MAXLEN ~ 10
但是，XTRIM旨在接受不同的修整政策，雖然現在隻實作了MAXLEN。鑒于這是一個明确的指令，将來有可能允許按時間來進行修整，因為以獨立的方式調用這個指令的使用者應該知道她或者他正在做什麼。
一個有用的驅逐政策是，XTRIM應該具有通過一系列ID删除的能力。目前這是不可能的，但在将來可能會實作，以便更友善地使用XRANGE 和 XTRIM來将Redis中的資料移到其他存儲系統中（如果需要）。
持久化，複制和消息安全性
與任何其他Redis資料結構一樣，Stream會異步複制到從節點，并持久化到AOF和RDB檔案中。但可能不那麼明顯的是，消費者組的完整狀态也會傳輸到AOF，RDB和從節點，是以如果消息在主節點是待處理的狀态，在從節點也會是相同的資訊。同樣，節點重新開機後，AOF檔案會恢複消費者組的狀态。
但是請注意，Redis Stream和消費者組使用Redis預設複制來進行持久化和複制，是以：
- 如果消息的持久性在您的應用程式中很重要，則AOF必須與強大的fsync政策一起使用。
- 預設情況下，異步複制不能保證複制XADD指令或者消費者組的狀态更改：在故障轉移後，可能會丢失某些内容，具體取決于從節點從主節點接收資料的能力。
- WAIT指令可以用于強制将更改傳輸到一組從節點上。但請注意，雖然這使得資料不太可能丢失，但由Sentinel或Redis群集運作的Redis故障轉移過程僅執行盡力檢查以故障轉移到最新的從節點，并且在某些特定故障下可能會選舉出缺少一些資料的從節點。 是以，在使用Redis Stream和消費者組設計應用程式時，確定了解你的應用程式在故障期間應具有的語義屬性，并進行相應地配置，評估它是否足夠安全地用于您的用例。
從Stream中删除單個項目
Stream還有一個特殊的指令可以通過ID從中間移除項目。一般來講，對于一個隻附加的資料結構來說，這也許看起來是一個奇怪的特征，但實際上它對于涉及例如隐私法規的應用程式是有用的。這個指令稱為XDEL，調用的時候隻需要傳遞Stream的名稱，在後面跟着需要删除的ID即可：
> XRANGE mystream - + COUNT 2
1) 1) 1526654999635-0
2) 1) "value"
2) "2"
2) 1) 1526655000369-0
2) 1) "value"
2) "3"
> XDEL mystream 1526654999635-0
(integer) 1
> XRANGE mystream - + COUNT 2
1) 1) 1526655000369-0
2) 1) "value"
2) "3"
但是在目前的實作中，在宏節點完全為空之前，記憶體并沒有真正回收，是以你不應該濫用這個特性。
零長度Stream
Stream與其他Redis資料結構有一個不同的地方在于，當其他資料結構沒有元素的時候，調用删除元素的指令會把key本身删掉。舉例來說就是，當調用ZREM指令将有序集合中的最後一個元素删除時，這個有序集合會被徹底删除。但Stream允許在沒有元素的時候仍然存在，不管是因為使用MAXLEN選項的時候指定了count為零（在XADD和XTRIM指令中），或者因為調用了XDEL指令。
存在這種不對稱性的原因是因為，Stream可能具有相關聯的消費者組，以及我們不希望因為Stream中沒有項目而丢失消費者組定義的狀态。目前，即使沒有相關聯的消費者組，Stream也不會被删除，但這在将來有可能會發生變化。
關于本文翻譯者
網名：eson
github： helloeson
打賞他（備注rediscn）
轉載于:https://www.cnblogs.com/williamjie/p/11201654.html
![PAT 1089 Insert or Merge[難][圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)