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是0，那么消費者組將會開始消費這個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，比如0或者任何其他有效的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也不會被刪除，但這在將來有可能會發生變化。