Redis Streams 介绍
Stream是Redis 5.0版本引入的一個新的數(shù)據(jù)類型，它以更抽象的方式模擬日志數(shù)據(jù)結(jié)構(gòu)，但日志仍然是完整的：就像一個日志文件，通常實現(xiàn)為以只附加模式打開的文件，Redis流主要是一個僅附加數(shù)據(jù)結(jié)構(gòu)。至少從概念上來講，因為Redis流是一種在內(nèi)存表示的抽象數(shù)據(jù)類型，他們實現(xiàn)了更加強大的操作，以此來克服日志文件本身的限制。
Stream是Redis的數(shù)據(jù)類型中最復(fù)雜的，盡管數(shù)據(jù)類型本身非常簡單，它實現(xiàn)了額外的非強制性的特性：提供了一組允許消費者以阻塞的方式等待生產(chǎn)者向Stream中發(fā)送的新消息，此外還有一個名為消費者組的概念。
消費者組最早是由名為Kafka（TM）的流行消息系統(tǒng)引入的。Redis用完全不同的術(shù)語重新實現(xiàn)了一個相似的概念，但目標是相同的：允許一組客戶端相互配合來消費同一個Stream的不同部分的消息。
Streams 基礎(chǔ)知識
為了理解Redis Stream是什么以及如何使用他們，我們將忽略所有的高級特性，從用于操縱和訪問它的命令方面來專注于數(shù)據(jù)結(jié)構(gòu)本身。這基本上是大多數(shù)其他Redis數(shù)據(jù)類型共有的部分，比如Lists，Sets，Sorted Sets等等。然而，需要注意的是Lists還有一個可選的更加復(fù)雜的阻塞API，由BLPOP等相似的命令導(dǎo)出。所以從這方面來說，Streams跟Lists并沒有太大的不同，只是附加的API更復(fù)雜、更強大。
因為Streams是只附加數(shù)據(jù)結(jié)構(gòu)，基本的寫命令，叫XADD，向指定的Stream追加一個新的條目。一個Stream條目不是簡單的字符串，而是由一個或多個鍵值對組成的。這樣一來，Stream的每一個條目就已經(jīng)是結(jié)構(gòu)化的，就像以CSV格式寫的只附加文件一樣，每一行由多個逗號割開的字段組成。
> XADD mystream * sensor-id 1234 temperature 19.8 1518951480106-0上面的例子中，調(diào)用了XADD命令往名為mystream的Stream中添加了一個條目sensor-id: 123, temperature: 19.8，使用了自動生成的條目ID，也就是命令返回的值，具體在這里是1518951480106-0。命令的第一個參數(shù)是key的名稱mystream，第二個參數(shù)是用于唯一確認Stream中每個條目的條目ID。然而，在這個例子中，我們傳入的參數(shù)值是*，因為我們希望由Redis服務(wù)器為我們自動生成一個新的ID。每一個新的ID都會單調(diào)增長，簡單來講就是，每次新添加的條目都會擁有一個比其它所有條目更大的ID。由服務(wù)器自動生成ID幾乎總是我們所想要的，需要顯式指定ID的情況非常少見。我們稍后會更深入地討論這個問題。實際上每個Stream條目擁有一個ID與日志文件具有另一種相似性，即使用行號或者文件中的字節(jié)偏移量來識別一個給定的條目。回到我們的XADD例子中，跟在key和ID后面的參數(shù)是組成我們的Stream條目的鍵值對。
使用XLEN命令來獲取一個Stream的條目數(shù)量：
> XLEN mystream (integer) 1條目 ID
條目ID由XADD命令返回，并且可以唯一的標識給定Stream中的每一個條目，由兩部分組成：
<millisecondsTime>-<sequenceNumber>毫秒時間部分實際是生成Stream ID的Redis節(jié)點的服務(wù)器本地時間，但是如果當前毫秒時間戳比以前的條目時間戳小的話，那么會使用以前的條目時間，所以即便是服務(wù)器時鐘向后跳，單調(diào)增長ID的特性仍然會保持不變。序列號用于以相同毫秒創(chuàng)建的條目。由于序列號是64位的，所以實際上對于在同一毫秒內(nèi)生成的條目數(shù)量是沒有限制的。
這樣的ID格式也許最初看起來有點奇怪，也許溫柔的讀者會好奇為什么時間會是ID的一部分。其實是因為Redis Streams支持按ID進行范圍查詢。由于ID與生成條目的時間相關(guān)，因此可以很容易地按時間范圍進行查詢。我們在后面講到XRANGE命令時，很快就能明白這一點。
如果由于某些原因，用戶需要與時間無關(guān)但實際上與另一個外部系統(tǒng)ID關(guān)聯(lián)的增量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中獲取數(shù)據(jù)
現(xiàn)在我們終于能夠通過XADD命令向我們的Stream中追加條目了。然而，雖然往Stream中追加數(shù)據(jù)非常明顯，但是為了提取數(shù)據(jù)而查詢Stream的方式并不是那么明顯，如果我們繼續(xù)使用日志文件進行類比，一種顯而易見的方式是模擬我們通常使用Unix命令tail -f來做的事情，也就是，我們可以開始監(jiān)聽以獲取追加到Stream的新消息。需要注意的是，不像Redis的阻塞列表，一個給定的元素只能到達某一個使用了冒泡風格的阻塞客戶端，比如使用類似BLPOP的命令，在Streams中我們希望看到的是多個消費者都能看到追加到Stream中的新消息，就像許多的tail -f進程能同時看到追加到日志文件的內(nèi)容一樣。用傳統(tǒng)術(shù)語來講就是我們希望Streams可以扇形分發(fā)消息到多個客戶端。
然而，這只是其中一種可能的訪問模式。我們還可以使用一種完全不同的方式來看待一個Stream：不是作為一個消息傳遞系統(tǒng)，而是作為一個時間序列存儲。在這種情況下，也許使附加新消息也非常有用，但是另一種自然查詢模式是通過時間范圍來獲取消息，或者使用一個游標來增量遍歷所有的歷史消息。這絕對是另一種有用的訪問模式。
最后，如果我們從消費者的角度來觀察一個Stream，我們也許想要以另外一種方式來訪問它，那就是，作為一個可以分區(qū)到多個處理此類消息的多個消費者的消息流，以便消費者組只能看到到達單個流的消息的子集。
Redis Streams通過不同的命令支持所有上面提到的三種訪問模式。接下來的部分將展示所有這些模式，從最簡單和更直接的使用：范圍查詢開始。
按范圍查詢: XRANGE 和 XREVRANGE
要根據(jù)范圍查詢Stream，我們只需要提供兩個ID，即start?和?end。返回的區(qū)間數(shù)據(jù)將會包括ID是start和end的元素，因此區(qū)間是完全包含的。兩個特殊的ID-?和?+分別表示可能的最小ID和最大ID。
> XRANGE mystream - + 1) 1) 1518951480106-02) 1) "sensor-id"2) "1234"3) "temperature"4) "19.8" 2) 1) 1518951482479-02) 1) "sensor-id"2) "9999"3) "temperature"4) "18.2"返回的每個條目都是有兩個元素的數(shù)組：ID和鍵值對列表。我們已經(jīng)說過條目ID與時間有關(guān)系，因為在字符-左邊的部分是創(chuàng)建Stream條目的本地節(jié)點上的Unix毫秒時間，即條目創(chuàng)建的那一刻（請注意：Streams的復(fù)制使用的是完全詳盡的XADD命令，因此從節(jié)點將具有與主節(jié)點相同的ID）。這意味著我可以使用XRANGE查詢一個時間范圍。然而為了做到這一點，我可能想要省略ID的序列號部分：如果省略，區(qū)間范圍的開始序列號將默認為0，結(jié)束部分的序列號默認是有效的最大序列號。這樣一來，僅使用兩個Unix毫秒時間去查詢，我們就可以得到在那段時間內(nèi)產(chǎn)生的所有條目（包含開始和結(jié)束）。例如，我可能想要查詢兩毫秒時間，可以這樣使用：
> XRANGE mystream 1518951480106 1518951480107 1) 1) 1518951480106-02) 1) "sensor-id"2) "1234"3) "temperature"4) "19.8"我在這個范圍內(nèi)只有一個條目，然而在實際數(shù)據(jù)集中，我可以查詢數(shù)小時的范圍，或者兩毫秒之間包含了許多的項目，返回的結(jié)果集很大。因此，XRANGE命令支持在最后放一個可選的COUNT選項。通過指定一個count，我可以只獲取前面N個項目。如果我想要更多，我可以拿返回的最后一個ID，在序列號部分加1，然后再次查詢。我們在下面的例子中看到這一點。我們開始使用XADD添加10個項目（我這里不具體展示，假設(shè)流mystream已經(jīng)填充了10個項目）。要開始我的迭代，每個命令只獲取2個項目，我從全范圍開始，但count是2。
> XRANGE mystream - + COUNT 2 1) 1) 1519073278252-02) 1) "foo"2) "value_1" 2) 1) 1519073279157-02) 1) "foo"2) "value_2"為了繼續(xù)下兩個項目的迭代，我必須選擇返回的最后一個ID，即1519073279157-0，并且在ID序列號部分加1。請注意，序列號是64位的，因此無需檢查溢出。在這個例子中，我們得到的結(jié)果ID是1519073279157-1，現(xiàn)在可以用作下一次XRANGE調(diào)用的新的start參數(shù)：
> XRANGE mystream 1519073279157-1 + COUNT 2 1) 1) 1519073280281-02) 1) "foo"2) "value_3" 2) 1) 1519073281432-02) 1) "foo"2) "value_4"依此類推。由于XRANGE的查找復(fù)雜度是O(log(N))，因此O(M)返回M個元素，這個命令在count較小時，具有對數(shù)時間復(fù)雜度，這意味著每一步迭代速度都很快。所以XRANGE也是事實上的流迭代器并且不需要XSCAN命令。
XREVRANGE命令與XRANGE相同，但是以相反的順序返回元素，因此XREVRANGE的實際用途是檢查一個Stream中的最后一項是什么：
> XREVRANGE mystream + - COUNT 1 1) 1) 1519073287312-02) 1) "foo"2) "value_10"請注意：XREVRANGE命令以相反的順序獲取start?和?stop參數(shù)。
使用XREAD監(jiān)聽新項目
當我們不想按照Stream中的某個范圍訪問項目時，我們通常想要的是訂閱到達Stream的新項目。這個概念可能與Redis中你訂閱頻道的Pub/Sub或者Redis的阻塞列表有關(guān)，在這里等待某一個key去獲取新的元素，但是這跟你消費Stream有著根本的不同：
提供監(jiān)聽到達Stream的新消息的能力的命令稱為XREAD。比XRANGE要更復(fù)雜一點，所以我們將從簡單的形式開始，稍后將提供整個命令布局。
> XREAD COUNT 2 STREAMS mystream 0 1) 1) "mystream"2) 1) 1) 1519073278252-02) 1) "foo"2) "value_1"2) 1) 1519073279157-02) 1) "foo"2) "value_2"以上是XREAD的非阻塞形式。注意COUNT選項并不是必需的，實際上這個命令唯一強制的選項是STREAMS，指定了一組key以及調(diào)用者已經(jīng)看到的每個Stream相應(yīng)的最大ID，以便該命令僅向客戶端提供ID大于我們指定ID的消息。
在上面的命令中，我們寫了STREAMS mystream 0，所以我們想要流?mystream中所有ID大于0-0的消息。正如你在上面的例子中所看到的，命令返回了鍵名，因為實際上可以通過傳入多個key來同時從不同的Stream中讀取數(shù)據(jù)。我可以寫一下，例如：STREAMS mystream otherstream 0 0。注意在STREAMS選項后面，我們需要提供鍵名稱，以及之后的ID。因此，STREAMS選項必須始終是最后一個。
除了XREAD可以同時訪問多個Stream這一事實，以及我們能夠指定我們擁有的最后一個ID來獲取之后的新消息，在個簡單的形式中，這個命令并沒有做什么跟XRANGE有太大區(qū)別的事情。然而，有趣的部分是我們可以通過指定BLOCK參數(shù)，輕松地將XREAD?變成一個?阻塞命令：
> XREAD BLOCK 0 STREAMS mystream $請注意，在上面的例子中，除了移除COUNT以外，我指定了新的BLOCK選項，超時時間為0毫秒（意味著永不超時）。此外，我并沒有給流?mystream傳入一個常規(guī)的ID，而是傳入了一個特殊的ID$。這個特殊的ID意思是XREAD應(yīng)該使用流?mystream已經(jīng)存儲的最大ID作為最后一個ID。以便我們僅接收從我們開始監(jiān)聽時間以后的新消息。這在某種程度上相似于Unix命令tail -f。
請注意當使用BLOCK選項時，我們不必使用特殊ID$。我們可以使用任意有效的ID。如果命令能夠立即處理我們的請求而不會阻塞，它將執(zhí)行此操作，否則它將阻止。通常如果我們想要從新的條目開始消費Stream，我們以$開始，接著繼續(xù)使用接收到的最后一條消息的ID來發(fā)起下一次請求，依此類推。
XREAD的阻塞形式同樣可以監(jiān)聽多個Stream，只需要指定多個鍵名即可。如果請求可以同步提供，因為至少有一個流的元素大于我們指定的相應(yīng)ID，則返回結(jié)果。否則，該命令將阻塞并將返回獲取新數(shù)據(jù)的第一個流的項目（根據(jù)提供的ID）。
跟阻塞列表的操作類似，從等待數(shù)據(jù)的客戶端角度來看，阻塞流讀取是公正的，由于語義是FIFO樣式。阻塞給定Stream的第一個客戶端是第一個在新項目可用時將被解除阻塞的客戶端。
XREAD命令沒有除了COUNT?和?BLOCK以外的其他選項，因此它是一個非常基本的命令，具有特定目的來攻擊消費者一個或多個流。使用消費者組API可以用更強大的功能來消費Stream，但是通過消費者組讀取是通過另外一個不同的命令來實現(xiàn)的，稱為XREADGROUP。本指南的下一節(jié)將介紹。
消費者組
當手頭的任務(wù)是從不同的客戶端消費同一個Stream，那么XREAD已經(jīng)提供了一種方式可以扇形分發(fā)到N個客戶端，還可以使用從節(jié)點來提供更多的讀取可伸縮性。然而，在某些問題中，我們想要做的不是向許多客戶端提供相同的消息流，而是從同一流向許多客戶端提供不同的消息子集。這很有用的一個明顯的例子是處理消息的速度很慢：能夠讓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使用了一個名為消費者組的概念。非常重要的一點是，從實現(xiàn)的角度來看，Redis的消費者組與Kafka (TM) 消費者組沒有任何關(guān)系，它們只是從實施的概念上來看比較相似，所以我決定不改變最初普及這種想法的軟件產(chǎn)品已有的術(shù)語。
消費者組就像一個偽消費者，從流中獲取數(shù)據(jù)，實際上為多個消費者提供服務(wù)，提供某些保證：
在某種程度上，消費者組可以被想象為關(guān)于Stream的一些狀態(tài)：
| 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) |如果你從這個視角來看，很容易理解一個消費者組能做什么，如何做到向給消費者提供他們的歷史待處理消息，以及當消費者請求新消息的時候，是如何做到只發(fā)送ID大于last_delivered_id的消息的。同時，如果你把消費者組看成Redis Stream的輔助數(shù)據(jù)結(jié)構(gòu)，很明顯單個Stream可以擁有多個消費者組，每個消費者組都有一組消費者。實際上，同一個Stream甚至可以通過XREAD讓客戶端在沒有消費者組的情況下讀取，同時有客戶端通過XREADGROUP在不同的消費者組中讀取。
現(xiàn)在是時候放大來查看基本的消費者組命令了，具體如下：
- XGROUP?用于創(chuàng)建，摧毀或者管理消費者組。
- XREADGROUP?用于通過消費者組從一個Stream中讀取。
- XACK?是允許消費者將待處理消息標記為已正確處理的命令。
創(chuàng)建一個消費者組
假設(shè)我已經(jīng)存在類型流的?mystream，為了創(chuàng)建消費者組，我只需要做：
> XGROUP CREATE mystream mygroup $ OK請注意：目前還不能為不存在的Stream創(chuàng)建消費者組，但有可能在不久的將來我們會給XGROUP命令增加一個選項，以便在這種場景下可以創(chuàng)建一個空的Stream。
如你所看到的上面這個命令，當創(chuàng)建一個消費者組的時候，我們必須指定一個ID，在這個例子中ID是$。這是必要的，因為消費者組在其他狀態(tài)中必須知道在第一個消費者連接時接下來要服務(wù)的消息，即消費者組創(chuàng)建完成時的最后消息ID是什么？如果我們就像上面例子一樣，提供一個$，那么只有從現(xiàn)在開始到達Stream的新消息才會被傳遞到消費者組中的消費者。如果我們指定的消息ID是0，那么消費者組將會開始消費這個Stream中的所有歷史消息。當然，你也可以指定任意其他有效的ID。你所知道的是，消費者組將開始傳遞ID大于你所指定的ID的消息。因為$表示Stream中當前最大ID的意思，指定$會有只消費新消息的效果。
現(xiàn)在消費者組創(chuàng)建好了，我們可以使用XREADGROUP命令立即開始嘗試通過消費者組讀取消息。我們會從消費者那里讀到，假設(shè)指定消費者分別是Alice和Bob，來看看系統(tǒng)會怎樣返回不同消息給Alice和Bob。
XREADGROUP和XREAD非常相似，并且提供了相同的BLOCK選項，除此以外還是一個同步命令。但是有一個強制的選項必須始終指定，那就是GROUP，并且有兩個參數(shù)：消費者組的名字，以及嘗試讀取的消費者的名字。選項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請注意：在這里消息是字段名稱，水果是關(guān)聯(lián)的值，記住Stream中的每一項都是小字典。
現(xiàn)在是時候嘗試使用消費者組讀取了：
> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream > 1) 1) "mystream"2) 1) 1) 1526569495631-02) 1) "message"2) "apple"XREADGROUP的響應(yīng)內(nèi)容就像XREAD一樣。但是請注意上面提供的GROUP <group-name> <consumer-name>，這表示我想要使用消費者組mygroup從Stream中讀取，我是消費者Alice。每次消費者使用消費者組中執(zhí)行操作時，都必須要指定可以這個消費者組中唯一標識它的名字。
在以上命令行中還有另外一個非常重要的細節(jié)，在強制選項STREAMS之后，鍵mystream請求的ID是特殊的ID?>。這個特殊的ID只在消費者組的上下文中有效，其意思是：消息到目前為止從未傳遞給其他消費者。
這幾乎總是你想要的，但是也可以指定一個真實的ID，比如0或者任何其他有效的ID，在這個例子中，我們請求XREADGROUP只提供給我們歷史待處理的消息，在這種情況下，將永遠不會在組中看到新消息。所以基本上XREADGROUP可以根據(jù)我們提供的ID有以下行為：
如果ID是特殊ID>，那么命令將會返回到目前為止從未傳遞給其他消費者的新消息，這有一個副作用，就是會更新消費者組的最后ID。 如果ID是任意其他有效的數(shù)字ID，那么命令將會讓我們訪問我們的歷史待處理消息。即傳遞給這個指定消費者（由提供的名稱標識）的消息集，并且到目前為止從未使用XACK進行確認。
我們可以立即測試此行為，指定ID為0，不帶任何COUNT選項：我們只會看到唯一的待處理消息，即關(guān)于apples的消息：
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0 1) 1) "mystream"2) 1) 1) 1526569495631-02) 1) "message"2) "apple"但是，如果我們確認這個消息已經(jīng)處理，它將不再是歷史待處理消息的一部分，因此系統(tǒng)將不再報告任何消息：
> XACK mystream mygroup 1526569495631-0 (integer) 1 > XREADGROUP GROUP mygroup Alice STREAMS mystream 0 1) 1) "mystream"2) (empty list or set)如果你還不清楚XACK是如何工作的，請不用擔心，這個概念只是已處理的消息不再是我們可以訪問的歷史記錄的一部分。
現(xiàn)在輪到Bob來讀取一些東西了：
> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream > 1) 1) "mystream"2) 1) 1) 1526569498055-02) 1) "message"2) "orange"2) 1) 1526569506935-02) 1) "message"2) "strawberry"Bob要求最多兩條消息，并通過同一消費者組mygroup讀取。所以發(fā)生的是Redis僅報告新消息。正如你所看到的，消息”apple”未被傳遞，因為它已經(jīng)被傳遞給Alice，所以Bob獲取到了orange和strawberry，以此類推。
這樣，Alice，Bob以及這個消費者組中的任何其他消費者，都可以從相同的Stream中讀取到不同的消息，讀取他們尚未處理的歷史消息，或者標記消息為已處理。這允許創(chuàng)建不同的拓撲和語義來從Stream中消費消息。
有幾件事需要記住：
- 消費者是在他們第一次被提及的時候自動創(chuàng)建的，不需要顯式創(chuàng)建。
- 即使使用XREADGROUP，你也可以同時從多個key中讀取，但是要讓其工作，你需要給每一個Stream創(chuàng)建一個名稱相同的消費者組。這并不是一個常見的需求，但是需要說明的是，這個功能在技術(shù)上是可以實現(xiàn)的。
- XREADGROUP命令是一個寫命令，因為當它從Stream中讀取消息時，消費者組被修改了，所以這個命令只能在master節(jié)點調(diào)用。
使用Ruby語言編寫的使用用戶組的消費者實現(xiàn)示例如下。 Ruby代碼的編寫方式，幾乎對使用任何其他語言編程的程序員或者不懂Ruby的人來說，都是清晰可讀的：
require 'redis'if ARGV.length == 0puts "Please specify a consumer name"exit 1 endConsumerName = ARGV[0] GroupName = "mygroup" r = Redis.newdef 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_backlogmyid = $lastidelsemyid = '>'enditems = r.xreadgroup('GROUP',GroupName,ConsumerName,'BLOCK','2000','COUNT','10','STREAMS',:my_stream_key,myid)if items == nilputs "Timeout!"nextend# 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 == 0items[0][1].each{|i|id,fields = i# Process the messageprocess_message(id,fields)# Acknowledge the message as processedr.xack(:my_stream_key,GroupName,id)$lastid = id} end正如你所看到的，這里的想法是開始消費歷史消息，即我們的待處理消息列表。這很有用，因為消費者可能已經(jīng)崩潰，因此在重新啟動時，我們想要重新讀取那些已經(jīng)傳遞給我們但還沒有確認的消息。通過這種方式，我們可以多次或者一次處理消息（至少在消費者失敗的場景中是這樣，但是這也受到Redis持久化和復(fù)制的限制，請參閱有關(guān)此主題的特定部分）。
消耗歷史消息后，我們將得到一個空的消息列表，我們可以切換到?>?，使用特殊ID來消費新消息。
從永久性失敗中恢復(fù)
上面的例子允許我們編寫多個消費者參與同一個消費者組，每個消費者獲取消息的一個子集進行處理，并且在故障恢復(fù)時重新讀取各自的待處理消息。然而在現(xiàn)實世界中，消費者有可能永久地失敗并且永遠無法恢復(fù)。由于任何原因停止后，消費者的待處理消息會發(fā)生什么呢？
Redis的消費者組提供了一個專門針對這種場景的特性，用以認領(lǐng)給定消費者的待處理消息，這樣一來，這些消息就會改變他們的所有者，并且被重新分配給其他消費者。這個特性是非常明確的，消費者必須檢查待處理消息列表，并且必須使用特殊命令來認領(lǐng)特定的消息，否則服務(wù)器將把待處理的消息永久分配給舊消費者，這樣不同的應(yīng)用程序就可以選擇是否使用這樣的特性，以及使用它的方式。
這個過程的第一步是使用一個叫做XPENDING的命令，這個命令提供消費者組中待處理條目的可觀察性。這是一個只讀命令，它總是可以安全地調(diào)用，不會改變?nèi)魏蜗⒌乃姓摺Ｔ谧詈唵蔚男问街?#xff0c;調(diào)用這個命令只需要兩個參數(shù)，即Stream的名稱和消費者組的名稱。
> XPENDING mystream mygroup 1) (integer) 2 2) 1526569498055-0 3) 1526569506935-0 4) 1) 1) "Bob"2) "2"當以這種方式調(diào)用的時候，命令只會輸出給定消費者組的待處理消息總數(shù)（在本例中是兩條消息），所有待處理消息中的最小和最大的ID，最后是消費者列表和每個消費者的待處理消息數(shù)量。我們只有Bob有兩條待處理消息，因為Alice請求的唯一一條消息已使用XACK確認了。
我們可以通過給XPENDING命令傳遞更多的參數(shù)來獲取更多信息，完整的命令簽名如下：
XPENDING <key> <groupname> [<start-id> <end-id> <count> [<conusmer-name>]]通過提供一個開始和結(jié)束ID（可以只是-和+，就像XRANGE一樣），以及一個控制命令返回的信息量的數(shù)字，我們可以了解有關(guān)待處理消息的更多信息。如果我們想要將輸出限制為僅針對給定使用者組的待處理消息，可以使用最后一個可選參數(shù)，即消費者組的名稱，但我們不會在以下示例中使用此功能。
> XPENDING mystream mygroup - + 10 1) 1) 1526569498055-02) "Bob"3) (integer) 741704584) (integer) 1 2) 1) 1526569506935-02) "Bob"3) (integer) 741704584) (integer) 1現(xiàn)在我們有了每一條消息的詳細信息：消息ID，消費者名稱，空閑時間（單位是毫秒，意思是：自上次將消息傳遞給某個消費者以來經(jīng)過了多少毫秒），以及每一條給定的消息被傳遞了多少次。我們有來自Bob的兩條消息，它們空閑了74170458毫秒，大概20個小時。
請注意，沒有人阻止我們檢查第一條消息內(nèi)容是什么，使用XRANGE即可。
> XRANGE mystream 1526569498055-0 1526569498055-0 1) 1) 1526569498055-02) 1) "message"2) "orange"我們只需要在參數(shù)中重復(fù)兩次相同的ID。現(xiàn)在我們有了一些想法，Alice可能會根據(jù)過了20個小時仍然沒有處理這些消息，來判斷Bob可能無法及時恢復(fù)，所以現(xiàn)在是時候認領(lǐng)這些消息，并繼續(xù)代替Bob處理了。為了做到這一點，我們使用XCLAIM命令。
這個命令非常的復(fù)雜，并且在其完整形式中有很多選項，因為它用于復(fù)制消費者組的更改，但我們只使用我們通常需要的參數(shù)。在這種情況下，它就像調(diào)用它一樣簡單：
XCLAIM <key> <group> <consumer> <min-idle-time> <ID-1> <ID-2> ... <ID-N>基本上我們說，對于這個特定的Stream和消費者組，我希望指定的ID的這些消息可以改變他們的所有者，并將被分配到指定的消費者<consumer>。但是，我們還提供了最小空閑時間，因此只有在上述消息的空閑時間大于指定的空閑時間時，操作才會起作用。這很有用，因為有可能兩個客戶端會同時嘗試認領(lǐng)一條消息：
Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0然而認領(lǐng)一條消息的副作用是會重置它的閑置時間！并將增加其傳遞次數(shù)的計數(shù)器，所以上面第二個客戶端的認領(lǐng)會失敗。通過這種方式，我們可以避免對消息進行簡單的重新處理（即使是在一般情況下，你仍然不能獲得準確的一次處理）。
下面是命令執(zhí)行的結(jié)果：
> XCLAIM mystream mygroup Alice 3600000 1526569498055-0 1) 1) 1526569498055-02) 1) "message"2) "orange"Alice成功認領(lǐng)了該消息，現(xiàn)在可以處理并確認消息，盡管原來的消費者還沒有恢復(fù)，也能往前推動。
從上面的例子很明顯能看到，作為成功認領(lǐng)了指定消息的副作用，XCLAIM命令也返回了消息數(shù)據(jù)本身。但這不是強制性的。可以使用JUSTID選項，以便僅返回成功認領(lǐng)的消息的ID。如果你想減少客戶端和服務(wù)器之間的帶寬使用量的話，以及考慮命令的性能，這會很有用，并且你不會對消息感興趣，因為稍后你的消費者的實現(xiàn)方式將不時地重新掃描歷史待處理消息。
認領(lǐng)也可以通過一個獨立的進程來實現(xiàn)：這個進程只負責檢查待處理消息列表，并將空閑的消息分配給看似活躍的消費者。可以通過Redis Stream的可觀察特性獲得活躍的消費者。這是下一個章節(jié)的主題。
消息認領(lǐng)及交付計數(shù)器
在XPENDING的輸出中，你所看到的計數(shù)器是每一條消息的交付次數(shù)。這樣的計數(shù)器以兩種方式遞增：消息通過XCLAIM成功認領(lǐng)時，或者調(diào)用XREADGROUP訪問歷史待處理消息時。
當出現(xiàn)故障時，消息被多次傳遞是很正常的，但最終它們通常會得到處理。但有時候處理特定的消息會出現(xiàn)問題，因為消息會以觸發(fā)處理代碼中的bug的方式被損壞或修改。在這種情況下，消費者處理這條特殊的消息會一直失敗。因為我們有傳遞嘗試的計數(shù)器，所以我們可以使用這個計數(shù)器來檢測由于某些原因根本無法處理的消息。所以一旦消息的傳遞計數(shù)器達到你給定的值，比較明智的做法是將這些消息放入另外一個Stream，并給系統(tǒng)管理員發(fā)送一條通知。這基本上是Redis Stream實現(xiàn)的dead letter概念的方式。
Streams 的可觀察性
缺乏可觀察性的消息系統(tǒng)很難處理。不知道誰在消費消息，哪些消息待處理，不知道給定Stream的活躍消費者組的集合，使得一切都不透明。因此，Redis Stream和消費者組都有不同的方式來觀察正在發(fā)生的事情。我們已經(jīng)介紹了XPENDING，它允許我們檢查在給定時刻正在處理的消息列表，以及它們的空閑時間和傳遞次數(shù)。
但是，我們可能希望做更多的事情，XINFO命令是一個可觀察性接口，可以與子命令一起使用，以獲取有關(guān)Stream或消費者組的信息。
這個命令使用子命令來顯示有關(guān)Stream和消費者組的狀態(tài)的不同信息，比如使用**XINFO STREAM?**可以報告關(guān)于Stream本身的信息。
> XINFO STREAM mystream1) length2) (integer) 133) radix-tree-keys4) (integer) 15) radix-tree-nodes6) (integer) 27) groups8) (integer) 29) first-entry 10) 1) 1524494395530-02) 1) "a"2) "1"3) "b"4) "2" 11) last-entry 12) 1) 1526569544280-02) 1) "message"2) "banana"輸出顯示了有關(guān)如何在內(nèi)部編碼Stream的信息，以及顯示了Stream的第一條和最后一條消息。另一個可用的信息是與這個Stream相關(guān)聯(lián)的消費者組的數(shù)量。我們可以進一步挖掘有關(guān)消費者組的更多信息。
> XINFO GROUPS mystream 1) 1) name2) "mygroup"3) consumers4) (integer) 25) pending6) (integer) 2 2) 1) name2) "some-other-group"3) consumers4) (integer) 15) pending6) (integer) 0正如你在這里和前面的輸出中看到的，XINFO命令輸出一系列鍵值對。因為這是一個可觀察性命令，允許人類用戶立即了解報告的信息，并允許命令通過添加更多字段來報告更多信息，而不會破壞與舊客戶端的兼容性。其他更高帶寬效率的命令，比如XPENDING，只報告沒有字段名稱的信息。
上面例子中的輸出（使用了子命令GROUPS）應(yīng)該能清楚地觀察字段名稱。我們可以通過檢查在此類消費者組中注冊的消費者，來更詳細地檢查特定消費者組的狀態(tài)。
> XINFO CONSUMERS mystream mygroup 1) 1) name2) "Alice"3) pending4) (integer) 15) idle6) (integer) 9104628 2) 1) name2) "Bob"3) pending4) (integer) 15) idle6) (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）分區(qū)的差異
Redis Stream的消費者組可能類似于基于Kafka（TM）分區(qū)的消費者組，但是要注意Redis Stream實際上非常不同。分區(qū)僅僅是邏輯的，并且消息只是放在一個Redis鍵中，因此不同客戶端的服務(wù)方式取決于誰準備處理新消息，而不是從哪個分區(qū)客戶端讀取。例如，如果消費者C3在某一點永久故障，Redis會繼續(xù)服務(wù)C1和C2，將新消息送達，就像現(xiàn)在只有兩個邏輯分區(qū)一樣。
類似地，如果一個給定的消費者在處理消息方面比其他消費者快很多，那么這個消費者在相同單位時間內(nèi)按比例會接收更多的消息。這是有可能的，因為Redis顯式地追蹤所有未確認的消息，并且記住了誰接收了哪些消息，以及第一條消息的ID從未傳遞給任何消費者。
但是，這也意味著在Redis中，如果你真的想把同一個Stream的消息分區(qū)到不同的Redis實例中，你必須使用多個key和一些分區(qū)系統(tǒng)，比如Redis集群或者特定應(yīng)用程序的分區(qū)系統(tǒng)。單個Redis Stream不會自動分區(qū)到多個實例上。
我們可以說，以下是正確的：
- 如果你使用一個Stream對應(yīng)一個消費者，則消息是按順序處理的。
- 如果你使用N個Stream對應(yīng)N個消費者，那么只有給定的消費者hits N個Stream的子集，你可以擴展上面的模型來實現(xiàn)。
- 如果你使用一個Stream對應(yīng)多個消費者，則對N個消費者進行負載平衡，但是在那種情況下，有關(guān)同一邏輯項的消息可能會無序消耗，因為給定的消費者處理消息3可能比另一個消費者處理消息4要快。
所以基本上Kafka分區(qū)更像是使用了N個不同的Redis鍵。而Redis消費者組是一個將給定Stream的消息負載均衡到N個不同消費者的服務(wù)端負載均衡系統(tǒng)。
設(shè)置Streams的上限
許多應(yīng)用并不希望將數(shù)據(jù)永久收集到一個Stream。有時在Stream中指定一個最大項目數(shù)很有用，之后一旦達到給定的大小，將數(shù)據(jù)從Redis中移到不那么快的非內(nèi)存存儲是有用的，適合用來記錄未來幾十年的歷史數(shù)據(jù)。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-02) 1) "value"2) "2" 2) 1) 1526655000369-02) 1) "value"2) "3"如果使用MAXLEN選項，當Stream的達到指定長度后，老的條目會自動被驅(qū)逐，因此Stream的大小是恒定的。目前還沒有選項讓Stream只保留給定數(shù)量的條目，因為為了一致地運行，這樣的命令必須為了驅(qū)逐條目而潛在地阻塞很長時間。比如可以想象一下如果存在插入尖峰，然后是長暫停，以及另一次插入，全都具有相同的最大時間。Stream會阻塞來驅(qū)逐在暫停期間變得太舊的數(shù)據(jù)。因此，用戶需要進行一些規(guī)劃并了解Stream所需的最大長度。此外，雖然Stream的長度與內(nèi)存使用是成正比的，但是按時間來縮減不太容易控制和預(yù)測：這取決于插入速率，該變量通常隨時間變化（當它不變化時，那么按尺寸縮減是微不足道的）。
然而使用MAXLEN進行修整可能很昂貴：Stream由宏節(jié)點表示為基數(shù)樹，以便非常節(jié)省內(nèi)存。改變由幾十個元素組成的單個宏節(jié)點不是最佳的。因此可以使用以下特殊形式提供命令：
XADD mystream MAXLEN ~ 1000 * ... entry fields here ...在選項MAXLEN和實際計數(shù)中間的參數(shù)~的意思是，我不是真的需要精確的1000個項目。它可以是1000或者1010或者1030，只要保證至少保存1000個項目就行。通過使用這個參數(shù)，僅當我們移除整個節(jié)點的時候才執(zhí)行修整。這使得命令更高效，而且這也是我們通常想要的。
還有XTRIM命令可用，它做的事情與上面講到的MAXLEN選項非常相似，但是這個命令不需要添加任何其他參數(shù)，可以以獨立的方式與Stream一起使用。
> XTRIM mystream MAXLEN 10或者，對于XADD選項：
> XTRIM mystream MAXLEN ~ 10但是，XTRIM旨在接受不同的修整策略，雖然現(xiàn)在只實現(xiàn)了MAXLEN。鑒于這是一個明確的命令，將來有可能允許按時間來進行修整，因為以獨立的方式調(diào)用這個命令的用戶應(yīng)該知道她或者他正在做什么。
一個有用的驅(qū)逐策略是，XTRIM應(yīng)該具有通過一系列ID刪除的能力。目前這是不可能的，但在將來可能會實現(xiàn)，以便更方便地使用XRANGE?和?XTRIM來將Redis中的數(shù)據(jù)移到其他存儲系統(tǒng)中（如果需要）。
持久化，復(fù)制和消息安全性
與任何其他Redis數(shù)據(jù)結(jié)構(gòu)一樣，Stream會異步復(fù)制到從節(jié)點，并持久化到AOF和RDB文件中。但可能不那么明顯的是，消費者組的完整狀態(tài)也會傳輸?shù)紸OF，RDB和從節(jié)點，因此如果消息在主節(jié)點是待處理的狀態(tài)，在從節(jié)點也會是相同的信息。同樣，節(jié)點重啟后，AOF文件會恢復(fù)消費者組的狀態(tài)。
但是請注意，Redis Stream和消費者組使用Redis默認復(fù)制來進行持久化和復(fù)制，所以：
- 如果消息的持久性在您的應(yīng)用程序中很重要，則AOF必須與強大的fsync策略一起使用。
- 默認情況下，異步復(fù)制不能保證復(fù)制XADD命令或者消費者組的狀態(tài)更改：在故障轉(zhuǎn)移后，可能會丟失某些內(nèi)容，具體取決于從節(jié)點從主節(jié)點接收數(shù)據(jù)的能力。
- WAIT命令可以用于強制將更改傳輸?shù)揭唤M從節(jié)點上。但請注意，雖然這使得數(shù)據(jù)不太可能丟失，但由Sentinel或Redis群集運行的Redis故障轉(zhuǎn)移過程僅執(zhí)行盡力檢查以故障轉(zhuǎn)移到最新的從節(jié)點，并且在某些特定故障下可能會選舉出缺少一些數(shù)據(jù)的從節(jié)點。 因此，在使用Redis Stream和消費者組設(shè)計應(yīng)用程序時，確保了解你的應(yīng)用程序在故障期間應(yīng)具有的語義屬性，并進行相應(yīng)地配置，評估它是否足夠安全地用于您的用例。
從Stream中刪除單個項目
Stream還有一個特殊的命令可以通過ID從中間移除項目。一般來講，對于一個只附加的數(shù)據(jù)結(jié)構(gòu)來說，這也許看起來是一個奇怪的特征，但實際上它對于涉及例如隱私法規(guī)的應(yīng)用程序是有用的。這個命令稱為XDEL，調(diào)用的時候只需要傳遞Stream的名稱，在后面跟著需要刪除的ID即可：
> XRANGE mystream - + COUNT 2 1) 1) 1526654999635-02) 1) "value"2) "2" 2) 1) 1526655000369-02) 1) "value"2) "3" > XDEL mystream 1526654999635-0 (integer) 1 > XRANGE mystream - + COUNT 2 1) 1) 1526655000369-02) 1) "value"2) "3"但是在當前的實現(xiàn)中，在宏節(jié)點完全為空之前，內(nèi)存并沒有真正回收，所以你不應(yīng)該濫用這個特性。
零長度Stream
Stream與其他Redis數(shù)據(jù)結(jié)構(gòu)有一個不同的地方在于，當其他數(shù)據(jù)結(jié)構(gòu)沒有元素的時候，調(diào)用刪除元素的命令會把key本身刪掉。舉例來說就是，當調(diào)用ZREM命令將有序集合中的最后一個元素刪除時，這個有序集合會被徹底刪除。但Stream允許在沒有元素的時候仍然存在，不管是因為使用MAXLEN選項的時候指定了count為零（在XADD和XTRIM命令中），或者因為調(diào)用了XDEL命令。
存在這種不對稱性的原因是因為，Stream可能具有相關(guān)聯(lián)的消費者組，以及我們不希望因為Stream中沒有項目而丟失消費者組定義的狀態(tài)。當前，即使沒有相關(guān)聯(lián)的消費者組，Stream也不會被刪除，但這在將來有可能會發(fā)生變化。
關(guān)于本文翻譯者
網(wǎng)名：eson
github：helloeson
總結(jié)
以上是生活随笔為你收集整理的Redis Streams 介绍的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Redis FAQ
- 下一篇： jvm理论-字节码指令