Redis Stream — nový abstrahovaný datový typ, který byl představen v Redis verze 5.0
Konceptuálně Redis Stream je List, do kterého můžete přidávat záznamy. Každý záznam má unikátní identifikátor. Standardně se identifikátor generuje automaticky a zahrnuje časovou značku. Můžete proto dotazovat rozsahy záznamů podle času nebo dostávat nová data, jakmile přicházejí do proudu, podobně jako Unix příkaz „tail -f“ čte logovací soubor a čeká na nová data. Všimněte si, že proud mohou poslouchat současně více klientů, stejně jako mnoho procesů „tail -f“ může současně číst soubor, aniž by došlo k jejich konfliktu.

Abychom porozuměli všem výhodám nového datového typu, připomeňme si rychle dlouho existující struktury Redis, které částečně opakují funkčnost Redis Stream.

Redis PUB/SUB

Redis Pub/Sub — jednoduchý systém zpráv, který je již zabudován ve vašem úložišti klíč-hodnota. Nicméně za jednoduchost se platí:

Pokud vydavatel z jakéhokoli důvodu selže, ztrácí všechny své odběratele
Vydavatel musí znát přesnou adresu všech svých odběratelů
Vydavatel může přetížit práci svých odběratelů, pokud jsou data publikována rychleji, než jsou zpracovávána
Zpráva je z vyrovnávací paměti vydavatele odstraněna ihned po publikaci, bez ohledu na to, kolika odběratelům byla doručena a jak rychle ji stihli zpracovat.
Všichni odběratelé obdrží zprávu současně. Odběratelé se musí nějak dohodnout na pořadí zpracování stejné zprávy.
Neexistuje vestavěný mechanismus potvrzení úspěšného zpracování zprávy odběratelem. Pokud odběratel obdrží zprávu a během zpracování selže, vydavatel o tom nebude vědět.

Redis List

Redis List — datová struktura, která podporuje příkazy pro čtení s blokací. Můžete přidávat a číst zprávy z začátku nebo konce seznamu. Na základě této struktury lze vytvořit slušný zásobník nebo frontu pro váš distribuovaný systém, což většinou stačí. Hlavní rozdíly oproti Redis Pub/Sub:

Zpráva je doručena jednomu klientovi. První klient, který se zablokoval při čtení, obdrží data jako první.
Clint musí sám iniciovat operaci čtení každé zprávy. List nic neví o klientech.
Zprávy jsou uchovávány, dokud je někdo nevyčte nebo je neodstraní explicitně. Pokud jste nastavili server Redis tak, aby ukládal data na disk, spolehlivost systému se výrazně zvyšuje.

Úvod do Streamu

Přidání záznamu do streamu

Tým XADD přidává nový záznam do streamu. Záznam není jen řetězec, ale skládá se z jedné nebo více dvojic klíč-hodnota. Tímto způsobem je každý záznam již strukturován a připomíná strukturu CSV souboru.

> XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0

V předchozím příkladu přidáváme do streamu s názvem (klíčem) „mystream“ dvě pole: „sensor-id“ a „temperature“ s hodnotami „1234“ a „19.8“. Jako druhý argument příkaz přijímá identifikátor, který bude přiřazen záznamu - tento identifikátor jednoznačně identifikuje každý záznam ve streamu. V tomto případě jsme však předali *, protože chceme, aby Redis vygeneroval nový identifikátor za nás. Každý nový identifikátor se bude zvyšovat. Proto každý nový záznam bude mít větší identifikátor ve srovnání s předchozími záznamy.

Formát identifikátoru

Identifikátor záznamu vrácený příkazem XADD, skládá se ze dvou částí:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix čas v milisekundách (čas serveru Redis). Pokud však aktuální čas vyjde jako stejný nebo menší než čas předchozího záznamu, použije se časová známka předchozího záznamu. Proto, pokud se čas serveru vrátí do minulosti, nový identifikátor si stále zachová vlastnost zvyšování.

sequenceNumber se používá pro záznamy vytvořené ve stejnou milisekundu. sequenceNumber bude zvýšen o 1 ve vztahu k předchozímu záznamu. Protože sequenceNumber má velikost 64 bitů, v praxi byste se neměli dostat k omezení počtu záznamů, které mohou být generovány během jedné milisekundy.

Formát těchto identifikátorů se na první pohled může zdát podivný. Nedůvěřivý čtenář se může ptát, proč je čas součástí identifikátoru. Důvodem je, že Redis streamy podporují dotazy na rozsahy podle identifikátorů. Vzhledem k tomu, že identifikátor je spojen s časem vytvoření záznamu, dává to možnost dotazovat se na časové rozsahy. Podíváme se na konkrétní příklad, když přejdeme k prozkoumání příkazu. XRANGE.

Pokud z jakéhokoli důvodu potřebuje uživatel zadat vlastní identifikátor, který je například spojen s nějakým externím systémem, můžeme jej předat příkazu XADD místo znaku * jak je uvedeno níže:

> XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2

Všimněte si, že v tomto případě si musíte sami hlídat zvyšování identifikátoru. V našem příkladu je minimální identifikátor „0-1“, takže příkaz nepřijme další identifikátor, který je roven nebo menší než „0-1“.

> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item

Počet záznamů ve streamu

Můžete získat počet záznamů ve streamu jednoduše pomocí příkazu XLEN. Pro náš příklad tento příkaz vrátí následující hodnotu:

> XLEN somestream
(integer) 2

Dotazy na rozsah — XRANGE a XREVRANGE

Když potřebujeme požádat o data v rozsahu, musíme uvést dva identifikátory — začátek a konec rozsahu. Vrácený rozsah bude zahrnovat všechny prvky, včetně hranic. Existují také dva speciální identifikátory „-“ a „+“, které znamenají nejmenší (první záznam) a největší (poslední záznam) identifikátor ve streamu. Následující příklad vrátí všechny záznamy streamu.

> 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"

Každý vrácený záznam představuje pole ze dvou prvků: identifikátor a seznam párů klíč-hodnota. Již jsme zmínili, že identifikátory záznamu mají vztah k času. Proto můžeme dotazovat rozsah určitého časového úseku. Můžeme však v dotazu uvést neúplný identifikátor, ale pouze Unix čas, přičemž část související s sequenceNumberOříznutá část identifikátoru se automaticky přiřadí k nule na začátku rozsahu a k maximálně možné hodnotě na konci rozsahu. Níže je uveden příklad, jak lze požádat o rozsah, který činí dvě milisekundy.

> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"

Máme pouze jeden záznam v tomto rozsahu, nicméně ve skutečných datových sadách může být vrácený výsledek obrovský. Z tohoto důvodu XRANGE podporuje možnost COUNT. Uvedením počtu můžeme jednoduše získat prvních N záznamů. Pokud potřebujeme získat další N záznamů (paginace), můžeme použít poslední získaný identifikátor, zvýšit jeho sequenceNumber o jednu a požádat znovu. Podívejme se na to v následujícím příkladu. Začneme přidávat 10 prvků pomocí XADD (předpokládejme, že tok mystream byl již naplněn 10 prvky). Abychom začali iteraci, získáváme po 2 prvcích na příkaz, začínáme s celým rozsahem, ale s COUNT rovným 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"

Abychom pokračovali v iteraci se dvěma následujícími prvky, musíme vybrat poslední získaný identifikátor, tedy 1519073279157-0, a přidat 1 k sequenceNumber.
výslednému identifikátoru, v tomto případě 1519073279157-1, který nyní lze použít jako nový argument začátku rozsahu pro další volání. XRANGE:

> 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"

A tak dále. Protože složitost XRANGE je O(log (N)) pro hledání a poté O(M) pro vrácení M prvků, každý krok iterace je rychlý. Tímto způsobem lze XRANGE efektivně iterovat toky.

Tým XREVRANGE je ekvivalentní XRANGE, ale vrací prvky v opačném pořadí:

> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
   2) 1) "foo"
      2) "value_10"

Mějte na paměti, že příkaz XREVRANGE přijímá argumenty rozsahu start a stop v opačném pořadí.

Čtení nových záznamů pomocí XREAD

Často se objeví úkol přihlásit se k toku a dostávat pouze nové zprávy. Tento koncept se může zdát podobný Redis Pub/Sub nebo blokujícímu Redis Listu, ale existují zásadní rozdíly v tom, jak používat Redis Stream:

Každá nová zpráva je výchozně doručena každému odběrateli. Toto chování se liší od blokujícího Redis List, kde novou zprávu přečte pouze jeden konkrétní odběratel.
Zatímco u Redis Pub/Sub se všechny zprávy zapomínají a nikdy se neuchovávají, ve Streamu se všechny zprávy uchovávají na neurčito (pokud klient explicitně nevyžádá jejich odstranění).
Redis Stream umožňuje rozdělený přístup k zprávám v rámci jednoho proudu. Konkrétní odběratel může vidět pouze svou osobní historii zpráv.

Můžete se přihlásit k proudu a dostávat nové zprávy pomocí příkazu XREAD. To je o něco složitější než XRANGE, takže nejprve začneme s jednoduššími příklady.

> 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"

V předchozím příkladu je uvedena neblokující forma XREAD. Všimněte si, že volba COUNT není povinná. Ve skutečnosti je jedinou povinnou volbou příkazu volba STREAMS, která určuje seznam proudů spolu s odpovídajícím maximálním identifikátorem. Napsali jsme „STREAMS mystream 0“ — chceme získat všechny záznamy proudu mystream s identifikátorem větším než „0-0“. Jak je vidět v příkladu, příkaz vrací název proudu, protože se můžeme přihlásit k několika proudům najednou. Mohli bychom napsat například „STREAMS mystream otherstream 0 0“. Všimněte si, že po volbě STREAMS musíme nejprve poskytnout názvy všech potřebných proudů a teprve poté seznam identifikátorů.

V této jednoduché formě příkaz nedělá nic zvláštního ve srovnání s XRANGE. Je však zajímavé, že můžeme snadno převést XREAD na blokující příkaz tím, že uvedeme argument BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
V uvedeném příkladu je uvedena nová volba BLOCK s časovým limitem 0 milisekund (to znamená nekonečné čekání). Navíc, místo běžného identifikátoru pro proud mystream byl předán speciální identifikátor $. Tento speciální identifikátor znamená, že XREAD by měl použít jako identifikátor maximální identifikátor v proudu mystream. Takže budeme dostávat pouze nové zprávy od okamžiku, kdy jsme začali poslouchat. V jistém smyslu to připomíná Unix příkaz „tail -f“.

Všimněte si, že při použití volby BLOCK není nutné používat speciální identifikátor $. Můžeme použít jakýkoli stávající identifikátor ve streamu. Pokud tým dokáže naši žádost okamžitě vyřídit bez zablokování, udělá to, v opačném případě se zablokuje.

Blokující XREAD může také současně sledovat více streamů, stačí uvést jejich názvy. V tomto případě příkaz vrátí záznam prvního streamu, do kterého byla data doručena. První odběratel zablokovaný pro daný stream obdrží data jako první.

Skupiny odběratelů

V určitých úlohách chceme omezit přístup odběratelů k zprávám uvnitř jednoho streamu. Příkladem, kdy to může být užitečné, je fronta zpráv s pracovníky, kteří obdrží různé zprávy ze streamu a umožní tak škálovat zpracování zpráv.

Pokud si představíme, že máme tři odběratele C1, C2, C3 a stream, který obsahuje zprávy 1, 2, 3, 4, 5, 6, 7, bude obsluha zpráv probíhat jako na diagramu níže:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
K dosažení tohoto efektu používá Redis Stream konceptu nazývaného Consumer Group. Tento koncept je podobný pseudo-odběrateli, který získává data ze streamu, ale ve skutečnosti je obsluhován několika odběrateli uvnitř skupiny, čímž poskytuje určité záruky:

Každá zpráva je doručena různým odběratelům uvnitř skupiny.
Uvnitř skupiny jsou odběratelé identifikováni podle jména, které je řetězcem s ohledem na velká a malá písmena. Pokud nějaký odběratel dočasně vypadne ze skupiny, může se do skupiny obnovit podle svého unikátního jména.
Každá Consumer Group dodržuje koncept „první nepřečtené zprávy“. Když odběratel požaduje nové zprávy, může obdržet pouze ty zprávy, které nikdy předtím nebyly doručeny žádnému odběrateli uvnitř skupiny.
Existuje příkaz explicitního potvrzení úspěšného zpracování zprávy odběratelem. Dokud nebude tento příkaz vyvolán, požadovaná zpráva zůstane ve stavu „čekající“.
Uvnitř Consumer Group může každý odběratel žádat o historii zpráv, které mu byly doručeny, ale ještě nebyly zpracovány (ve stavu „čekající“).

V jistém smyslu lze stav skupiny prezentovat takto:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" s nevyřízenými zprávami  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" s nevyřízenými zprávami 
|       ... (a tak dále)                             
+----------------------------------------+

Nyní je čas seznámit se se základními příkazy pro Consumer Group, konkrétně:

XGROUP se používá pro vytváření, ničení a správu skupin
XREADGROUP se používá pro čtení proudu přes skupinu
XACK — je příkaz, který umožňuje odběrateli označit zprávu jako úspěšně zpracovanou

Vytvoření Consumer Group

Předpokládejme, že proud mystream již existuje. Příkaz pro vytvoření skupiny tedy bude vypadat takto:

> XGROUP CREATE mystream mygroup $ OK
Při vytváření skupiny musíme předat identifikátor, od kterého bude skupina přijímat zprávy. Pokud chceme prostě dostávat všechny nové zprávy, můžeme použít speciální identifikátor $ (jak v našem výše uvedeném příkladu). Pokud místo speciálního identifikátoru uvedeme 0, bude skupině zpřístupněna všechna zpráva proudu.

Nyní, když byla skupina vytvořena, můžeme okamžitě začít číst zprávy pomocí příkazu XREADGROUP. Tento příkaz je velmi podobný příkazu XREAD a podporuje volitelnou možnost BLOCK. Existuje však povinná možnost GROUP, která musí být vždy uvedena s dvěma argumenty: názvem skupiny a názvem odběratele. Také se podporuje možnost COUNT.

Než začneme číst proud, pojďme do něj umístit několik zpráv:

> 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

A nyní se pokusíme přečíst tento proud přes skupinu:

> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

Výše uvedený příkaz doslovně zní takto:

„Já, Alice-odběratel, člen skupiny mygroup, chci přečíst ze proudu mystream jednu zprávu, která dosud nebyla nikomu doručena.“

Při každé operaci s skupinou musí účastník uvést své jméno, aby se jednoznačně identifikoval uvnitř skupiny. V předchozím příkazu je ještě jeden velmi důležitý detail — speciální identifikátor „>“. Tento speciální identifikátor filtruje zprávy a zanechává pouze ty, které ještě nebyly doručeny.

Také v některých případech můžete uvést skutečný identifikátor, jako je 0 nebo jakýkoli jiný platný identifikátor. V tomto případě příkaz XREADGROUP vám vrátí historii zpráv se stavem „pending“, které byly doručeny uvedenému účastníkovi (Alice), ale dosud nebyly potvrzeny příkazem XACK.

Můžeme toto chování ověřit ihned uvedením identifikátoru 0, bez volby COUNT. Uvidíme pouze jednu čekající zprávu, tedy zprávu se jménem jablko:

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

Pokud však potvrdíme zprávu jako úspěšně zpracovanou, již se nezobrazí:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (prázdný seznam nebo množina)

Teď je řada na Bobovi, aby si něco přečetl:

> 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, člen skupiny mygroup, požádal o maximálně dvě zprávy. Příkaz hlásí pouze nedoručené zprávy kvůli speciálnímu identifikátoru „>“. Jak vidíte, zpráva „apple“ se nezobrazuje, protože již byla doručena Alici, takže Bob dostává „orange“ a „strawberry“.

Takto mohou Alice, Bob a jakýkoli jiný účastník skupiny číst různé zprávy ze stejného proudu. Mohou také číst svou historii neošetřených zpráv nebo označovat zprávy jako zpracované.

Existuje několik věcí, na které je třeba myslet:

Jakmile účastník přečte zprávu příkazem XREADGROUP, tato zpráva přechází do stavu „pending“ a je přiřazena tomuto konkrétnímu účastníkovi. Ostatní účastníci skupiny nebudou moci tuto zprávu přečíst.
Účastníci se automaticky vytvářejí při prvním zmínění, není nutné je explicitně vytvářet.
Pomocí XREADGROUP Můžete číst zprávy z několika různých proudů současně, ale abyste to zvládli, musíte předem vytvořit skupiny se stejným názvem pro každý proud pomocí XGROUP

Obnovení po havárii

Odběratel se může zotavit po havárii a znovu si přečíst svůj seznam zpráv se statusem „čekající“. V reálném světě však mohou odběratelé nakonec selhat úplně. Co se stane s visícími zprávami odběratele, pokud se mu nepodaří obnovit po havárii?
Consumer Group nabízí funkci, která se používá přesně pro takové situace — když je potřeba změnit vlastníka zpráv.

Nejdříve je potřeba zavolat příkaz XPENDING, který zobrazí všechny zprávy skupiny se statusem „čekající“. V nejjednodušší formě je příkaz volán pouze se dvěma argumenty: názvem proudu a názvem skupiny:

> XPENDING mystream mygroup
1) (integer) 2
2) 1526569498055-0
3) 1526569506935-0
4) 1) 1) "Bob"
      2) "2"

Příkaz vrátil počet nezpracovaných zpráv pro celou skupinu a pro každého odběratele. Máme pouze Boba se dvěma nezpracovanými zprávami, protože jediná zpráva, kterou požadovala Alice, byla potvrzena pomocí XACK.

Můžeme požádat o další informace, používající více argumentů:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — rozsah identifikátorů (je možné použít „-“ a „+“)
{count} — počet pokusů o doručení
{consumer-name} — jméno skupiny

> 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

Teď máme detaily pro každou zprávu: identifikátor, jméno odběratele, čas prostoj v milisekundách a nakonec počet pokusů o doručení. Máme dvě zprávy od Boba, které prostojí 74170458 milisekund, což je přibližně 20 hodin.

Všimněte si, že nám nic nebrání zkontrolovat, jaký byl obsah zprávy, jednoduše pomocí XRANGE.

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

Musíme jen opakovat stejný identifikátor dvakrát v argumentech. Teď, když máme nějakou představu, může se Alice rozhodnout, že po 20 hodinách prostojí Bob pravděpodobně nezotaví, a je čas požádat o tyto zprávy a obnovit jejich zpracování místo Boba. Pro to použijeme příkaz XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Pomocí tohoto příkazu můžeme získat "cizí" zprávu, která ještě nebyla zpracována, tím, že změníme vlastníka na {consumer}. Můžeme však také poskytnout minimální dobu nečinnosti {min-idle-time}. To pomáhá předejít situaci, kdy se dva klienti snaží současně změnit vlastníka u stejné zprávy.

Klient 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Klient 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
První klient resetuje dobu nečinnosti a zvýší počítač doručení. Takže druhý klient ji nebude moci požádat.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

Zpráva byla úspěšně vyžádána Alicí, která ji nyní může zpracovat a potvrdit.

Z výše uvedeného příkladu je vidět, že úspěšné provedení požadavku vrací obsah samotné zprávy. Není to však nutné. Možnost JUSTID může být použita k vrácení pouze identifikátorů zprávy. To je užitečné, pokud vás nezajímají detaily zprávy a chcete zvýšit výkon systému.

Počítadlo doručení

Počítadlo, které vidíte v výstupu XPENDING — to je počet doručení každé zprávy. Toto počítadlo se zvyšuje dvěma způsoby: když je zpráva úspěšně vyžádána prostřednictvím XCLAIM nebo když je použito volání XREADGROUP.

Je normální, že některé zprávy jsou doručovány několikrát. Hlavní je, aby v konečném důsledku byly všechny zprávy zpracovány. Někdy při zpracování zprávy nastanou problémy kvůli poškození samotné zprávy nebo způsobuje zpracování zprávy chybu v kódu zpracovatele. V takovém případě se může stát, že tuto zprávu nikdo nebude schopen zpracovat. Vzhledem k tomu, že máme počítadlo pokusů o doručení, můžeme toto počítadlo využít k odhalení takových situací. Proto, jakmile počítadlo doručení dosáhne zadaného velkého čísla, pravděpodobně bude rozumnější umístit tuto zprávu do jiného proudu a poslat upozornění systémovému administrátorovi.

Stav proudů

Tým XINFO se používá k dotazování různých informací o proudu a jeho skupinách. Například základní tvar příkazu vypadá takto:

> 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"

Příkaz výše zobrazuje obecné informace o zadaném streamu. Teď trochu složitější příklad:

> XINFO GROUPS mystream
1) 1) název
   2) "mygroup"
   3) spotřebitelé
   4) (celé číslo) 2
   5) čekající
   6) (celé číslo) 2
2) 1) název
   2) "some-other-group"
   3) spotřebitelé
   4) (celé číslo) 1
   5) čekající
   6) (celé číslo) 0

Příkaz výše zobrazuje obecné informace o všech skupinách zadaného streamu.

> XINFO CONSUMERS mystream mygroup
1) 1) název
   2) "Alice"
   3) čekající
   4) (celé číslo) 1
   5) nečinný
   6) (celé číslo) 9104628
2) 1) název
   2) "Bob"
   3) čekající
   4) (celé číslo) 1
   5) nečinný
   6) (celé číslo) 83841983

Příkaz výše zobrazuje informace o všech odběratelích zadaného streamu a skupiny.
Pokud zapomenete syntaxi příkazu, jednoduše se obraťte na samotný příkaz pro pomoc:

> XINFO HELP
1) XINFO {podpříkaz} arg arg ... arg. Podpříkazy jsou:
2) CONSUMERS {klíč} {název skupiny}  -- Zobrazit spotřebitelské skupiny skupiny {název skupiny}.
3) GROUPS {klíč}                 -- Zobrazit spotřebitelské skupiny streamu.
4) STREAM {klíč}                 -- Zobrazit informace o streamu.
5) HELP                         -- Vytisknout tuto pomoc.

Omezení velikosti streamu

Mnoho aplikací nechce shromažďovat data ve streamu navždy. Často je užitečné mít maximální povolený počet zpráv ve streamu. V jiných případech je užitečné přesunout všechny zprávy ze streamu do jiného trvalého úložiště, když je dosaženo dané velikosti streamu. Velikost streamu můžete omezit pomocí parametru MAXLEN v příkazu. XADD:

> 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
(celé číslo) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

Při použití MAXLEN se staré záznamy automaticky odstraňují, když je dosaženo uvedené délky, takže má stream stále konstantní velikost. Nicméně truncation v tomto případě není nejefektivnější způsob v paměti Redis. Situaci lze zlepšit následujícími způsoby:

XADD mystream MAXLEN ~ 1000 * ... pole pro položky zde ...
Argument ~ v uvedeném příkladu znamená, že nemusíme omezovat délku streamu na konkrétní hodnotu. V našem příkladu to může být jakékoli číslo větší nebo rovno 1000 (například 1000, 1010 nebo 1030). Jen jsme jasně uvedli, že chceme, aby náš stream uchoval alespoň 1000 záznamů. To činí práci s pamětí mnohem efektivnější uvnitř Redis.

Existuje také samostatný příkaz XTRIM, který provádí to samé:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Trvalé ukládání a replikace

Redis Stream se asynchronně replikuje na slave uzly a ukládá se do souborů typu AOF (snapshot všech dat) a RDB (log všech zápisových operací). Podporuje také replikaci stavu Consumer Groups. Proto, pokud je zpráva ve stavu „pending“ na master uzlu, na slave uzlech bude mít tato zpráva stejný stav.

Odstranění jednotlivých prvků z proudu

Pro odstranění zpráv existuje speciální příkaz XDEL. Příkaz přijímá název proudu, za kterým následují identifikátory zpráv, které je třeba odstranit:

> 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"

Při použití tohoto příkazu je třeba vzít v úvahu, že paměť nebude skutečně uvolněna okamžitě.

Proudy nulové délky

Rozdíl mezi proudy a jinými datovými strukturami Redis spočívá v tom, že když jiné datové struktury již nemají prvky uvnitř, jako vedlejší efekt, samotná datová struktura bude odstraněna z paměti. Například, seřazená množina bude zcela odstraněna, když se zavolá ZREM pro odstranění posledního prvku. Na druhou stranu, proudům je dovoleno zůstat v paměti, i když nemají žádný prvek uvnitř.