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