Redis Stream — novi apstraktni tip podataka predstavljen u Redis-u s verzijom 5.0
Konceptualno, Redis Stream je Lista u koju možete dodavati zapise. Svaki zapis ima jedinstveni identifikator. Po defaultu, identifikator se automatski generira i uključuje vremensku oznaku. Dakle, možete upititi opsege zapisa prema vremenu ili dobivati nove podatke dok pristižu u stream, poput Unix komande „tail -f“ koja čita logfile i čeka nove podatke. Imajte na umu da stream može slušati više klijenata istovremeno, kao što mnogi procesi „tail -f“ mogu istovremeno čitati datoteku bez sukoba.

Da bismo razumjeli sve prednosti novog tipa podataka, hajde da se uskratko prisjetimo starih struktura Redis-a koje djelomično ponavljaju funkcionalnost Redis Stream-a.

Redis PUB/SUB

Redis Pub/Sub — jednostavan sustav poruka već ugrađen u vaše key-value skladište. Međutim, zbog jednostavnosti dolazi se do sljedećih nedostataka:

Ako izdavač iz bilo kojeg razloga nestane, gubi sve svoje pretplatnike.
Izdavač mora znati točnu adresu svih svojih pretplatnika.
Izdavač može preopteretiti svoje pretplatnike ako se podaci objavljuju brže nego što ih oni mogu obraditi.
Poruka se uklanja iz buffer-a izdavača odmah nakon objavljivanja, bez obzira na to koliko je pretplatnika primilo poruku i koliko su brzo uspjeli obraditi tu poruku.
Svi pretplatnici dobit će poruku istovremeno. Pretplatnici se sami moraju dogovoriti o redoslijedu obrade iste poruke.
Ne postoji ugrađeni mehanizam za potvrdu uspješne obrade poruke od strane pretplatnika. Ako pretplatnik primi poruku, a tijekom obrade padne, izdavač neće saznati za to.

Redis Lista

Redis Lista — struktura podataka koja podržava komande čitanja s blokadom. Možete dodavati i čitati poruke s početka ili kraja liste. Na temelju ove strukture možete napraviti dobar stek ili red za vaš distribuirani sustav i to će u većini slučajeva biti dovoljno. Glavne razlike od Redis Pub/Sub:

Poruka se isporučuje jednom klijentu. Prvi klijent koji je blokiran na čitanju će prvi primiti podatke.
Clint mora sam inicirati operaciju čitanja svake poruke. List ništa ne zna o klijentima.
Poruke se čuvaju dok ih netko ne pročita ili ne izbriše eksplicitno. Ako ste postavili Redis poslužitelj da zapisuje podatke na disk, pouzdanost sustava značajno se povećava.

Uvod u Stream

Dodavanje zapisa u tok

Komanda XADD dodaje novi zapis u tok. Zapis nije samo niz, on se sastoji od jednog ili više parova ključ-vrijednost. Tako je svaki zapis već strukturiran i nalikuje strukturi CSV datoteke.

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

U gornjem primjeru dodajemo u tok pod imenom (ključem) «mystream» dva polja: «sensor-id» i «temperature» s vrijednostima «1234» i «19.8» odnosno. Kao drugi argument, komanda prima identifikator koji će biti dodijeljen zapisu — taj identifikator nedvosmisleno identificira svaki zapis u toku. Međutim, u ovom slučaju proslijedili smo *, jer želimo da Redis generira novi identifikator za nas. Svaki novi identifikator će se povećavati. Stoga će svaki novi zapis imati veći identifikator u odnosu na prethodne zapise.

Format identifikatora

Identifikator zapisa, koji vraća komanda XADD, sastoji se od dviju dijelova:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix vrijeme u milisekundama (vrijeme poslužitelja Redis). Međutim, ako se trenutna vrijeme pokaže istim ili manjim od vremena prethodnog zapisa, koristi se vremenska oznaka prethodnog zapisa. Stoga, ako vrijeme poslužitelja vraća u prošlost, novi identifikator još uvijek će zadržati svoju osobinu povećanja.

sequenceNumber koristi se za zapise koji su kreirani u istoj milisekundi. sequenceNumber bit će povećan za 1 u odnosu na prethodni zapis. Budući da sequenceNumber ima veličinu od 64 bita, u praksi ne biste trebali naići na ograničenje u broju zapisa koji se mogu generirati tijekom jedne milisekunde.

Format takvih identifikatora na prvi pogled može izgledati čudno. Nepovjerljivi čitatelj se može zapitati zašto je vrijeme dio identifikatora. Razlog je u tome što Redis streamovi podržavaju upite raspona po identifikatorima. Budući da je identifikator povezan s vremenom stvaranja zapisa, to omogućava upite po vremenskim rasponima. Razmotrit ćemo konkretan primjer kada pređemo na istraživanje naredbe XRANGE.

Ako iz nekog razloga korisnik treba odrediti vlastiti identifikator, koji je, na primjer, povezan s nekom vanjskom sustavom, možemo ga proslijediti naredbi XADD umjesto znaka * kao što je prikazano u nastavku:

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

Imajte na umu da u ovom slučaju morate sami pratiti povećanje identifikatora. U našem primjeru minimalni identifikator iznosi „0-1“, stoga naredba neće prihvatiti još jedan identifikator koji je jednak ili manji od „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

Broj zapisa u streamu

Možete dobiti broj zapisa u streamu jednostavno koristeći naredbu XLEN. Za naš primjer ova naredba će vratiti sljedeću vrijednost:

> XLEN somestream
(integer) 2

Upiti po rasponu — XRANGE i XREVRANGE

Da bismo zatražili podatke po rasponu, trebamo navesti dva identifikatora — početak i kraj raspona. Vraćeni raspon će uključivati sve elemente, uključujući granice. Također postoje dva posebna identifikatora „-” i „+”, koji znače najmanji (prvi zapis) i najveći (posljednji zapis) identifikator u streamu. Primjer u nastavku ispisat će sve zapise streama.

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

Svaka vraćena stavka predstavlja niz od dva elementa: identifikator i popis parova ključ-vrijednost. Već smo spomenuli da identifikatori zapisa imaju veze s vremenom. Stoga možemo zatražiti raspon određenog vremenskog intervala. Međutim, možemo u upitu navesti ne potpuni identifikator, već samo Unix vrijeme, izostavljajući dio koji se odnosi na sequenceNumberOdređeni dio identifikatora automatski će se postaviti na nulu na početku opsega i na maksimalnu moguću vrijednost na kraju opsega. Evo primjera kako možete zatražiti opseg koji iznosi dvije milisekunde.

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

Imamo samo jedan zapis u ovom opsegu, međutim, u stvarnim skupovima podataka vraćeni rezultat može biti ogroman. Zbog toga XRANGE podržava opciju COUNT. Ako odredimo količinu, možemo jednostavno dobiti prvih N zapisa. Ako trebamo dobiti sljedećih N zapisa (paginacija), možemo koristiti posljednji dobiven identifikator, povećati ga sequenceNumber za jedan i ponovno zatražiti. Pogledajmo to u sljedećem primjeru. Počinjemo dodavati 10 elemenata pomoću XADD (pretpostavimo da je stream mystream već bio ispunjen s 10 elemenata). Da bismo započeli iteraciju, dobivajući 2 elementa po naredbi, počinjemo s punim opsegom, ali s COUNT postavljenim na 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"

Da bismo nastavili s iteracijom sljedeća dva elementa, trebamo odabrati posljednji dobiven identifikator, tj. 1519073279157-0, i dodati 1 na sequenceNumber.
Rezultantni identifikator, u ovom slučaju 1519073279157-1, sada se može koristiti kao novi argument za početak opsega za sljedeći poziv 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"

I tako dalje. Budući da složenost XRANGE iznosi O(log (N)) za pretraživanje, a zatim O(M) za vraćanje M elemenata, svaki korak iteracije je brz. Tako, putem XRANGE možete učinkovito iterirati kroz tokove.

Komanda XREVRANGE je ekvivalent XRANGE, ali vraća elemente u obrnutom redoslijedu:

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

Napomena: naredba XREVRANGE prima argumente opsega start i stop u obrnutom redoslijedu.

Čitanje novih zapisa pomoću XREAD

Često se javlja potreba da se pretplatite na tok i primate samo nove poruke. Ova koncepcija može izgledati slična Redis Pub/Sub ili blokirajućem Redis Listu, ali postoje bitne razlike u tome kako koristiti Redis Stream:

Svaka nova poruka zadano se isporučuje svakom pretplatniku. Ovo ponašanje se razlikuje od blokirajuće Redis Liste, gdje će nova poruka biti pročitana samo od strane jednog pretplatnika.
Dok se u Redis Pub/Sub sve poruke zaboravljaju i nikada se ne čuvaju, u Streamu sve poruke se čuvaju na neodređeno vrijeme (osim ako klijent eksplicitno ne zatraži brisanje).
Redis Stream omogućuje razdvajanje pristupa porukama unutar jednog toka. Konkretnog pretplatnika može vidjeti samo svoju osobnu povijest poruka.

Možete se pretplatiti na tok i primati nove poruke koristeći naredbu XREAD. Ovo je malo složenije od XRANGE, stoga ćemo prvo započeti s jednostavnijim primjerima.

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

U gornjem primjeru navedena je neblokirajuća forma XREAD. Imajte na umu da opcija COUNT nije obavezna. Zapravo, jedina obavezna opcija naredbe je opcija STREAMS, koja postavlja popis tokova zajedno s odgovarajućim maksimalnim identifikatorom. Napisali smo «STREAMS mystream 0» — želimo primiti sve unose toka mystream s identifikatorom većim od «0-0». Kao što se može vidjeti iz primjera, naredba vraća ime toka, jer se možemo pretplatiti na više tokova istovremeno. Mogli bismo napisati, na primjer, «STREAMS mystream otherstream 0 0». Imajte na umu da nakon opcije STREAMS prvo trebamo navesti nazive svih potrebnih tokova, a tek potom popis identifikatora.

U ovoj jednostavnoj formi naredba ne radi ništa posebno u odnosu na XRANGE. Međutim, zanimljivo je to što možemo lako pretvoriti XREAD u blokirajuću naredbu, navodeći argument BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
U gornjem primjeru navedena je nova opcija BLOCK s vremenom čekanja od 0 milisekundi (to znači beskonačno čekanje). Štoviše, umjesto uobičajenog identifikatora za tok mystream, proslijeđen je posebni identifikator $. Ovaj posebni identifikator znači da XREAD treba koristiti kao identifikator maksimalni identifikator u toku mystream. Tako ćemo primiti samo nove poruke, počinjući od trenutka kada smo počeli slušati. Na neki način, to je slično Unix naredbi «tail -f».

Napomena da prilikom korištenja opcije BLOCK nije potrebno koristiti posebni identifikator $. Možemo koristiti bilo koji postojeći identifikator u toku. Ako tim može odmah obraditi naš zahtjev bez blokiranja, učinit će to, inače će se blokirati.

Blokirajući XREAD također može istovremeno slušati više tokova, samo je potrebno navesti njihova imena. U tom slučaju tim će vratiti zapis prvog toka u koji su podaci stigli. Prvi pretplatnik blokiran za taj tok dobit će podatke prvi.

Grupice potrošača

U određenim zadacima želimo ograničiti pristup pretplatnika porukama unutar jednog toka. Primjer kada to može biti korisno je red poruka s radnicima koji će primati različite poruke toka, omogućujući skaliranje obrade poruka.

Ako zamislimo da imamo tri pretplatnika C1, C2, C3 i tok koji sadrži poruke 1, 2, 3, 4, 5, 6, 7, onda će obrada poruka izgledati kao na dijagramu ispod:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Kako bismo postigli ovaj efekt, Redis Stream koristi koncept koji se naziva Grupa potrošača. Ovaj koncept je sličan pseudo-pretplatniku koji prima podatke iz toka, ali ga zapravo opslužuju više pretplatnika unutar grupe, pružajući određene garancije:

Svaka poruka se dostavlja različitim pretplatnicima unutar grupe.
Unutar grupe, pretplatnici se identificiraju po imenu, koje je niz sa osjetljivim na velika i mala slova. Ako neki pretplatnik privremeno ispadne iz grupe, može se ponovno priključiti grupi koristeći svoje jedinstveno ime.
Svaka Grupa potrošača slijedi koncept „prve nepročitane poruke“. Kada pretplatnik zatraži nove poruke, može dobiti samo one poruke koje nikada prije nisu bile isporučene nijednom pretplatniku unutar grupe.
Postoji naredba za izričito potvrđivanje uspješnog obrade poruke od strane pretplatnika. Dok ne bude pozvana ova naredba, zatražena poruka će ostati u statusu „pending“.
Unutar Grupe potrošača svaki pretplatnik može zatražiti povijest poruka koje su mu isporučene, ali još nisu obrađene (u statusu „pending“).

Na neki način, stanje grupe može se predstaviti ovako:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" s neisporučenim porukama  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" s neisporučenim porukama 
|       ... (i tako dalje)                             
+----------------------------------------+

Sada je došlo vreme da se upoznamo s osnovnim komandama za grupu potrošača, a to su:

XGROUP koristi se za kreiranje, uništavanje i upravljanje grupama
XREADGROUP koristi se za čitanje toka kroz grupu
XACK — ova komanda omogućava pretplatniku da označi poruku kao uspešno obrađenu

Kreiranje grupe potrošača

Pretpostavimo da tok mystream već postoji. Tada će komanda za kreiranje grupe izgledati ovako:

> XGROUP CREATE mystream mygroup $ OK
Prilikom kreiranja grupe, moramo navesti identifikator s kojeg će grupa primati poruke. Ako želimo samo da primamo sve nove poruke, možemo koristiti posebni identifikator $ (kao u našem gornjem primeru). Ako umesto posebnog identifikatora navedemo 0, grupi će biti dostupne sve poruke toka.

Sada, kada je grupa kreirana, možemo odmah početi čitati poruke koristeći komandu XREADGROUP. Ova komanda je vrlo slična XREAD i podržava opciju BLOCK. Međutim, postoji obavezna opcija GROUP koja mora da se navede sa dva argumenta: ime grupe i ime pretplatnika. Opcija COUNT se takođe podržava.

Pre nego što počnemo da čitamo tok, hajde da tamo stavimo nekoliko poruka:

> 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 sada pokušajmo da pročitajmo ovaj tok kroz grupu:

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

Gornja komanda u prevodu glasi:

«Ja, pretplatnik Alisa, član grupe mygroup, želim da pročitam iz toka mystream jednu poruku koja nikome nije ranije isporučena.»

Svaki put kada pretplatnik izvrši operaciju s grupom, mora navesti svoje ime kako bi se jedinstveno identificirao unutar grupe. U gornjoj naredbi postoji još jedna vrlo važna stavka — posebni identifikator „>“. Ovaj posebni identifikator filtrira poruke, ostavljajući samo one koje do sada nisu bile dostavljene.

Također, u posebnim slučajevima, možete navesti stvarni identifikator, poput 0 ili bilo kojeg drugog valjanog identifikatora. U tom slučaju, naredba XREADGROUP vratit će vam povijest poruka sa statusom „pending“, koje su bile dostavljene određenom pretplatniku (Alice), ali još nisu potvrđene naredbom XACK.

Možemo provjeriti ovo ponašanje odmah navodeći identifikator 0, bez opcije COUNT. Jednostavno ćemo vidjeti jedinstvenu poruku koja čeka, tj. poruku s jabukom:

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

Međutim, ako potvrdimo poruku kao uspješno obrađenu, više se neće prikazivati:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)

Sada je red na Bobu da nešto pročita:

> 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, član grupe mygroup, zatražio je ne više od dvije poruke. Naredba izvještava samo o neispunjenim porukama zbog posebnog identifikatora „>“. Kao što vidite, poruka „apple“ se ne prikazuje, jer je već dostavljena Alici, tako da Bob dobiva „orange“ i „strawberry“.

Tako, Alice, Bob i bilo koji drugi pretplatnik grupe mogu čitati različite poruke iz istog toka. Također mogu čitati svoju povijest neobrađenih poruka ili označiti poruke kao obrađene.

Postoji nekoliko stvari koje treba imati na umu:

Jednom kada pretplatnik čita poruku pomoću naredbe XREADGROUP, ta poruka prelazi u stanje „pending“ i pridružuje se tom konkretno pretplatniku. Ostali pretplatnici grupe neće moći pročitati tu poruku.
Pretplatnici se automatski stvaraju pri prvom spominjanju, nema potrebe za njihovim eksplicitnim stvaranjem.
Pomoću XREADGROUP možete čitati poruke iz više različitih tokova istovremeno, međutim, da bi to funkcioniralo, potrebno je unaprijed stvoriti grupe s istim imenom za svaki tok pomoću XGROUP

Obnova nakon kvara

Pretplatnik se može oporaviti nakon kvara i ponovno pročitati svoj popis poruka sa statusom „pending“. Međutim, u stvarnom svijetu pretplatnici mogu trajno propasti. Što se događa s suspendiranim porukama pretplatnika ako se nije uspio oporaviti od kvara?
Consumer Group nudi funkciju koja se koristi upravo u takvim situacijama — kada je potrebno promijeniti vlasnika poruka.

Prvo morate pozvati naredbu XPENDING, koja prikazuje sve poruke grupe sa statusom „pending“. U svom najjednostavnijem obliku, naredba se poziva samo s dva argumenta: imenom toka i imenom grupe:

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

Naredba je prikazala broj neobrađenih poruka za cijelu grupu i za svakog pretplatnika. Imamo samo Boba s dvije neobrađene poruke, jer je jedina poruka koju je zatražila Alisa potvrđena pomoću XACK.

Možemo zatražiti dodatne informacije koristeći više argumenata:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — raspon identifikatora (može se koristiti „-” i „+”)
{count} — broj pokušaja isporuke
{consumer-name} — ime grupe

> 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

Sada imamo detalje za svaku poruku: identifikator, ime pretplatnika, vrijeme zastoja u milisekundama i, konačno, broj pokušaja isporuke. Imamo dvije poruke od Boba, a one stoje već 74170458 milisekundi, otprilike 20 sati.

Imajte na umu da nam ništa ne sprečava da provjerimo kakav je bio sadržaj poruke, jednostavno koristeći XRANGE.

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

Samo trebamo dvaput ponoviti isti identifikator u argumentima. Sada, kada imamo neku ideju, Alisa može odlučiti da, nakon 20 sati zastoja, Bob vjerojatno neće uspjeti i da je došlo vrijeme da zatraži te poruke i nastavi njihovu obradu umjesto Boba. Za to koristimo naredbu XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Pomoću ove komande možemo dobiti "tuđu" poruku koja još nije obrađena, promjenom vlasnika na {consumer}. Međutim, možemo također osigurati minimalno vrijeme neaktivnosti {min-idle-time}. To pomaže izbjeći situaciju kada dvoje klijenata pokušava istovremeno promijeniti vlasnika istih poruka:

Klijent 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Klijent 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Prvi klijent će resetirati vrijeme neaktivnosti i povećati broj dostava. Tako drugi klijent neće moći zatražiti tu poruku.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "poruka"
      2) "naranča"

Poruka je uspješno zatražena od strane Alice, koja sada može obraditi poruku i potvrditi je.

Iz gornjeg primjera možemo vidjeti da uspješna realizacija zahtjeva vraća sadržaj same poruke. Međutim, to nije obavezno. Opcija JUSTID može se koristiti za vraćanje samo identifikatora poruke. To je korisno ako niste zainteresirani za detalje poruke i želite povećati performanse sustava.

Brojač dostava

Brojač koji vidite u izlazu XPENDING — je broj dostava svake poruke. Ovaj brojač se povećava na dva načina: kada se poruka uspješno zatraži putem XCLAIM ili kada se koristi poziv XREADGROUP.

Normalno je da se neke poruke isporučuju nekoliko puta. Najvažnije je da su na kraju sve poruke obrađene. Ponekad se mogu pojaviti problemi tijekom obrade poruke zbog oštećenja same poruke ili obradna logika uzrokuje grešku u kodu rukovatelja. U takvim slučajevima, može se dogoditi da nitko ne bude u mogućnosti obraditi tu poruku. Budući da imamo brojač pokušaja dostave, možemo iskoristiti ovaj brojač za otkrivanje takvih situacija. Stoga, kada brojač dostava dostigne određeni veliki broj koji ste postavili, vjerojatno bi bilo razborito smjestiti tu poruku u drugi tok i poslati obavijest sistemskom administratoru.

Stanje tokova

Komanda XINFO koristi se za upit raznih informacija o toku i njegovim grupama. Na primjer, osnovni oblik komande izgleda ovako:

> 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) "poruka"
       2) "banana"

Gornja komanda prikazuje opće informacije o navedenom streamu. Sada malo složeniji primjer:

> XINFO GROUPS mystream
1) 1) ime
   2) "mygroup"
   3) potrošači
   4) (cijeli broj) 2
   5) neobrađeni
   6) (cijeli broj) 2
2) 1) ime
   2) "some-other-group"
   3) potrošači
   4) (cijeli broj) 1
   5) neobrađeni
   6) (cijeli broj) 0

Gornja komanda prikazuje opće informacije o svim grupama navedenog streama.

> XINFO CONSUMERS mystream mygroup
1) 1) ime
   2) "Alice"
   3) neobrađeni
   4) (cijeli broj) 1
   5) neaktivan
   6) (cijeli broj) 9104628
2) 1) ime
   2) "Bob"
   3) neobrađeni
   4) (cijeli broj) 1
   5) neaktivan
   6) (cijeli broj) 83841983

Gornja komanda prikazuje informacije o svim pretplatnicima navedenog streama i grupe.
Ako zaboravite sintaksu komande, jednostavno potražite pomoć u samoj komandi:

> XINFO HELP
1) XINFO {podkomanda} arg arg ... arg. Podkomande su:
2) CONSUMERS {ključ} {ime grupe}  -- Prikaži potrošačke grupe grupe {ime grupe}.
3) GROUPS {ključ}                 -- Prikaži potrošačke grupe streama.
4) STREAM {ključ}                 -- Prikaži informacije o streamu.
5) HELP                         -- Ispiši ovu pomoć.

Ograničenje veličine streama

Mnoge aplikacije ne žele sakupljati podatke u stream beskojnice. Često je korisno imati maksimalni broj poruka u streamu. U drugim slučajevima korisno je prebaciti sve poruke iz streama u drugo trajno skladište kada se postigne zadata veličina streama. Veličinu streama moguće je ograničiti pomoću parametra MAXLEN u komandi 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
(cijeli broj) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

Pri korištenju MAXLEN, stari zapisi automatski se brišu kada se postigne određena dužina, tako da stream ima stalnu veličinu. Međutim, obrezivanje se u ovom slučaju ne odvija na najproduktivniji način u memoriji Redis-a. Situaciju možete poboljšati na sljedeći način:

XADD mystream MAXLEN ~ 1000 * ... polja unosa ovdje ...
Argument ~ u gornjem primjeru označava da nije nužno ograničiti dužinu streama određenom vrijednošću. U našem primjeru to može biti bilo koji broj veći ili jednak 1000 (na primjer, 1000, 1010 ili 1030). Jednostavno smo jasno naznačili da želimo da naš stream pohrani barem 1000 zapisa. To čini rad s memorijom mnogo učinkovitijim unutar Redis-a.

Također postoji posebna komanda XTRIM, koja izvršava isto:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Trajna pohrana i replikacija

Redis Stream asinkrono se replicira na slave čvorove i pohranjuje se u datoteke tipa AOF (snapshot svih podataka) i RDB (log svih operacija pisanja). Također se podržava replikacija stanja Consumer Groups. Stoga, ako je poruka u statusu „pending“ na master čvoru, na slave čvorovima ta će poruka imati isti status.

Brisanje pojedinih elemenata iz toka

Za brisanje poruka postoji posebna naredba XDEL. Naredba prima ime toka, nakon čega slijede identifikatori poruka koje treba izbrisati:

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

Pri korištenju ove naredbe treba imati na umu da će se zapravo memorija osloboditi ne odmah.

Tokovi nulte duljine

Razlika između tokova i drugih struktura podataka u Redis-u je u tome što, kada druge strukture podataka više nemaju elemenata unutar sebe, kao sporedni učinak, sama struktura podataka će biti uklonjena iz memorije. Na primjer, sortirani skup će se potpuno ukloniti kada poziv ZREM ukloni posljednji element. Umjesto toga, tokovima je dopušteno da ostanu u memoriji, čak i kada nemaju niti jedan element unutar.