Redis Stream — nov tip podatkov, predstavljen v Redis z izidom različice 5.0
Konceptualno je Redis Stream List, v katerega lahko dodajate zapise. Vsak zapis ima edinstveno identifikacijsko številko. Po privzetku se identifikator ustvari samodejno in vključuje časovno žig. Zato lahko zahtevate obsege zapisov po času ali pridobite nove podatke v trenutku, ko pritečejo v tok, tako kot Unixova ukazna „tail -f“, ki bere dnevniške datoteke in čaka na nove podatke. Upoštevajte, da lahko tok hkrati posluša več odjemalcev, kot lahko številni procesi „tail -f“ hkrati berejo datoteko, ne da bi si pri tem nasprotovali.

Da bi razumeli vse prednosti novega tipa podatkov, si poglejmo obstoječe strukture Redis, ki delno ponavljajo funkcionalnost Redis Stream.

Redis PUB/SUB

Redis Pub/Sub — preprost sistem sporočil, ki je že vgrajen v vaše key-value skladišče. Toda za preprostost je potrebno plačati:

Če založnik iz kakršnega koli razloga preneha delovati, izgubi vse svoje naročnike.
Založnik mora poznati točno mesto vseh svojih naročnikov.
Založnik lahko obremenjuje delo svojih naročnikov, če so podatki objavljeni hitreje, kot jih ti obdelujejo.
Sporočilo se odstrani iz založnikovega predpomnilnika takoj po objavi, ne glede na to, koliko naročnikov ga je prejelo in kako hitro so ga uspeli obdelati.
Vsi naročniki bodo sporočilo prejeli hkrati. Naročniki se morajo nekako dogovoriti o vrstnem redu obdelave istega sporočila.
Ne obstaja vgrajeni mehanizem za potrditev uspešne obdelave sporočila s strani naročnika. Če naročnik prejme sporočilo in med obdelavo odpove, založnik o tem ne izve.

Redis List

Redis List — struktura podatkov, ki podpira ukaze branja z zaklepanjem. Lahko dodajate in preberete sporočila z začetka ali konca seznama. Na osnovi te strukture lahko ustvarite dober sklad ali čakalno vrsto za vaš porazdeljeni sistem in to bo v večini primerov dovolj. Glavne razlike od Redis Pub/Sub:

Sporočilo se dostavi enemu odjemalcu. Prvi odjemalec, ki je z zaklepanjem prebral, bo prejel podatke prvi.
Clint mora sam inicirati operaciju čitanja svake poruke. List ništa ne zna o klijentima.
Poruke se čuvaju dok ih neko ne pročita ili ih ne izbriše eksplicitno. Ako ste podesili Redis server da pritisne podatke na disk, pouzdanost sistema naglo raste.

Uvod u Stream

Dodavanje zapisa u stream

Komanda XADD dodaje novi zapis u stream. Zapis nije samo niz, sastoji se od jednog ili više parova ključ-vrednost. Na taj način, svaki zapis je već strukturiran i podseća na strukturu CSV datoteke.

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

U gornjem primeru dodajemo u stream nazvan (ključ) "mystream" dva polja: "sensor-id" i "temperature" sa vrednostima "1234" i "19.8" respektivno. Kao drugi argument, komanda prihvata identifikator koji će biti dodeljen zapisu — ovaj identifikator jedinstveno identifikuje svaki zapis u streamu. Međutim, u ovom slučaju smo preneli *, jer želimo da Redis generiše novi identifikator umesto nas. Svaki novi identifikator će biti povećan. Zato svaki novi zapis će imati veći identifikator u odnosu na prethodne zapise.

Format identifikatora

Identifikator zapisa, koji vraća komanda XADD, sastoji se od dva dela:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix vreme u milisekundama (vreme strežnika Redis). Međutim, ako se trenutno vreme ispostavi da je isto ili manje od vremena prethodnog zapisa, koristi se vremenska oznaka prethodnog zapisa. Tako da, ako vreme servera ide unazad, novi identifikator će i dalje zadržati osobinu povećanja.

sequenceNumber se koristi za zapise kreirane u istoj milisekundi. sequenceNumber biće povećan za 1 u odnosu na prethodni zapis. Pošto sequenceNumber ima veličinu od 64 bita, praktično ne bi trebalo da se suočite sa ograničenjem na broj zapisa koji mogu biti generisani tokom jedne milisekunde.

Format takih identifikatorjev se na prvi pogled morda zdi nenavaden. Nezaupljiv bralec se lahko sprašuje, zakaj je čas del identifikatorja. Razlog je, da tokovi Redis podpirajo poizvedbe v razponu po identifikatorjih. Ker je identifikator povezan s časom ustvarjanja zapisa, to omogoča poizvedovanje po časovnih razponih. Oglejšemo si konkretni primer, ko se bomo lotili preučevanja ukaza XRANGE.

Če uporabniku iz kakršnegakoli razloga zadeva zahteva določen identifikator, ki je na primer povezan z neko zunanjo sistemom, ga lahko posredujemo ukazu XADD namesto znaka * kot je prikazano spodaj:

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

Upoštevajte, da morate v tem primeru sami skrbeti za povečanje identifikatorja. V našem primeru je minimalni identifikator '0-1', zato ukaz ne bo sprejel drugega identifikatorja, ki je enak ali manjši 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

Število zapisov v toku

Število zapisov v toku lahko dobimo kar z uporabo ukaza XLEN. Za naš primer bo ta ukaz vrnil naslednjo vrednost:

> XLEN somestream
(integer) 2

Poizvedbe po razponu — XRANGE in XREVRANGE

Da bi poizvedli podatke po razponu, moramo navesti dva identifikatorja — začetek in konec razpona. Vrnjeni razpon bo vključil vse elemente, vključno z mejami. Obstajata tudi dva posebna identifikatorja '-' in '+', ki ustrezno pomenita najmanjši (prvi zapis) in največji (zadnji zapis) identifikator v toku. Spodnji primer bo izpisal vse zapise toka.

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

Vsak vrnjeni zapis predstavlja niz dveh elementov: identifikator in seznam parov ključ-vrednost. Že smo omenili, da so identifikatorji zapisov povezani s časom. Zato lahko poizvedujemo po razponu določenega časovnega obdobja. Vendar pa lahko v poizvedbi navedemo ne popoln identifikator, temveč le Unix čas, tako da del, ki se nanaša na sequenceNumberIzpuščeni del identifikatorja se samodejno obravnava kot nič na začetku razpona in kot največja mogoča vrednost na koncu razpona. Spodaj je primer, kako lahko zahtevate razpon, ki znaša dve milisekundi.

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

Imamo samo en zapis v tem razponu, vendar so lahko v resničnih naborih podatkov vrnjeni rezultati izjemno veliki. Iz tega razloga XRANGE podpira možnost COUNT. Z določitvijo števila lahko preprosto pridobimo prvih N zapisov. Če želimo pridobiti naslednjih N zapisov (paginacija), lahko uporabimo zadnji prejeti identifikator, mu dodamo sequenceNumber ena in ponovno zahtevamo. Oglejmo si to v naslednjem primeru. Začnemo dodajati 10 elementov z uporabo XADD (predpostavimo, da je bil tok mystream že napolnjen z 10 elementi). Da začnemo iteracijo, pri čemer pridobivamo po 2 elementa na ukaz, začnemo z celotnim razponom, vendar s COUNT, ki je enaka 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 bi nadaljevali iteracijo s naslednjima dvema elementoma, moramo izbrati zadnji prejeti identifikator, torej 1519073279157-0, in dodati 1 k sequenceNumber.
rezultantnemu identifikatorju, v tem primeru 1519073279157-1, zdaj ga lahko uporabimo kot nov argument začetnega razpona za naslednji klic 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"

In tako naprej. Ker je kompleksnost XRANGE O(log (N)) za iskanje in nato O(M) za vračilo M elementov, je vsak korak iteracije hiter. Tako je s XRANGE možno učinkovito iterirati potoke.

Komanda XREVRANGE je ekvivalentno XRANGE, vendar vrne elemente v obratnem vrstnem redu:

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

Upoštevajte, da ukaz XREVRANGE sprejema argumente razpona start in stop v obratnem vrstnem redu.

Branje novih zapisov z uporabo XREAD

Pogosto se pojavi naloga naročiti se na tok in prejemati samo nova sporočila. Ta koncept se lahko zdi podoben Redis Pub/Sub ali blokirajočemu Redis Listu, vendar obstajajo bistvene razlike v tem, kako uporabiti Redis Stream:

Vsaka nova sporočila so privzeto dostavljena vsakemu naročniku. To vedenje se razlikuje od blokirajoče Redis Liste, kjer bo novo sporočilo prebral le en sam naročnik.
Medtem ko v Redis Pub/Sub vsa sporočila izginejo in nikoli niso shranjena, se v Stream vsa sporočila shranijo za nedoločen čas (razen če stranka izrecno zahteva njihovo izbris).
Redis Stream omogoča ločevanje dostopa do sporočil znotraj enega toka. Konkreten naročnik lahko vidi le svojo osebno zgodovino sporočil.

Lahko se naročite na tok in prejemate nova sporočila z uporabo ukaza XREAD. To je nekoliko bolj zapleteno kot XRANGE, zato bomo najprej začeli s preprostimi primeri.

> 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 zgornjem primeru je navedena neblokirajoča oblika XREAD. Upoštevajte, da ni obvezna možnost COUNT. Dejansko je edina obvezna možnost ukaza možnost STREAMS, ki določa seznam tokov skupaj s pripadajočim največjim identifikatorjem. Napisali smo „STREAMS mystream 0“ — želimo prejeti vse zapise toka mystream z identifikatorjem večjim od „0-0“. Kot je razvidno iz primera, ukaz vrne ime toka, saj se lahko naročimo na več tokov hkrati. Lahko bi napisali, na primer, „STREAMS mystream otherstream 0 0“. Upoštevajte, da moramo po možnosti STREAMS najprej zagotoviti imena vseh potrebnih tokov in šele nato seznam identifikatorjev.

V tej preprosti obliki ukaz ne naredi nič posebnega v primerjavi z XRANGE. Vendar je zanimivo, da lahko zlahka spremenimo XREAD v blokirajoči ukaz, tako da navedemo argument BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
V zgornjem primeru je navedena nova možnost BLOCK s časom čakanja 0 milisekund (to pomeni neskončno čakanje). Poleg tega je bil namesto običajnega identifikatorja za tok mystream posredovan poseben identifikator $. Ta poseben identifikator pomeni, da XREAD mora uporabiti največji identifikator v toku mystream. Tako bomo prejemali le nova sporočila, ki se pričela od trenutka, ko smo začeli poslušati. V nekem smislu je to podobno unix ukazu „tail -f“.

Upoštevajte, da pri uporabi možnosti BLOCK ni treba uporabljati posebnega identifikatorja $. Uporabimo lahko katerikoli obstoječ identifikator v potoku. Če bo ekipa lahko takoj obdelala našo zahtevo brez blokiranja, to stori, sicer se bo blokirala.

Blokirajoč XREAD lahko tudi posluša več potokov hkrati, le da je potrebno navesti njihova imena. V tem primeru bo ekipa vrnila zapis prvega potoka, v katerega so prišli podatki. Prvi naročnik, blokiran za ta potok, bo prejel podatke kot prvi.

Potrošniške skupine

Pri določenih nalogah želimo omejiti dostop naročnikov do sporočil znotraj enega potoka. Primer, ko je to lahko koristno, je vrstni red sporočil s delavci, ki bodo prejemali različna sporočila iz potoka, kar omogoča skaliranje obdelave sporočil.

Če si predstavljamo, da imamo tri naročnike C1, C2, C3 in potok, ki vsebuje sporočila 1, 2, 3, 4, 5, 6, 7, bo obdelava sporočil potekala tako, kot je prikazano na spodnjem diagramu:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Da bi dosegli ta učinek, Redis Stream uporablja koncept, imenovan Consumer Group. Ta koncept je podoben pseudo-naročniku, ki prejema podatke iz potoka, vendar ga dejansko obdelujejo več naročnikov znotraj skupine, kar zagotavlja določena jamstva:

Vsako sporočilo je dostavljeno različnim naročnikom znotraj skupine.
Znotraj skupine so naročniki identificirani po imenu, ki je niz, občutljiv na velike in male črke. Če nek naročnik začasno izpade iz skupine, se lahko vrne v skupino po svojem edinstvenem imenu.
Vsaka potrošniška skupina sledi konceptu 'prvo neprebrano sporočilo'. Ko naročnik zahteva nova sporočila, lahko prejme samo tista sporočila, ki nikoli prej niso bila dostavljena nobenemu naročniku v skupini.
Obstaja ukaz za izrecno potrditev uspešnega obvladovanja sporočila s strani naročnika. Dokler ta ukaz ni poklican, bo zahtevano sporočilo ostalo v statusu 'pending'.
Znotraj potrošniške skupine lahko vsak naročnik zahteva zgodovino sporočil, ki so bila dostavljena prav njemu, vendar še niso bila obdelana (v statusu 'pending').

V nekem smislu se stanje skupine lahko prikaže tako:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" s čakajočimi sporočili  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" s čakajočimi sporočili 
|       ... (in tako naprej)                             
+----------------------------------------+

Zdaj je čas, da se seznanimo z osnovnimi ukazi za Consumer Group, in sicer:

XGROUP se uporablja za ustvarjanje, uničenje in upravljanje skupin
XREADGROUP se uporablja za branje toka prek skupine
XACK — ta ukaz omogoča prejemniku, da označi sporočilo kot uspešno obdelano

Ustvarjanje Consumer Group

Predpostavimo, da tok mystream že obstaja. Takrat bo ukaz za ustvarjanje skupine videti takole:

> XGROUP CREATE mystream mygroup $ OK
Ko ustvarjamo skupino, moramo posredovati identifikator, s katerega bo skupina začela prejemati sporočila. Če želimo prejemati le vsa nova sporočila, lahko uporabimo poseben identifikator $ (kot v našem zgornjem primeru). Če namesto posebnega identifikatorja navedemo 0, bodo skupini na voljo vsa sporočila toka.

Zdaj, ko je skupina ustvarjena, lahko takoj začnemo brati sporočila s pomočjo ukaza XREADGROUP. Ta ukaz je zelo podoben XREAD in podpira neobvezno možnost BLOCK. Vendar pa je tu obvezna možnost GROUP, ki jo je vedno treba navesti z dvema argumentoma: ime skupine in ime prejemnika. Možnost COUNT je prav tako podprta.

Preden začnemo brati tok, dajmo vanj nekaj sporočil:

> 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

Sedaj poskusimo prebrati ta tok skozi skupino:

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

Zgornji ukaz dobesedno pravi naslednje:

„Jaz, Alice-prejemnik, član skupine mygroup, želim brati iz toka mystream eno sporočilo, ki nikoli ni bilo nikomur dostavljeno prej.“

Vsakač, ko naročnik izvede operacijo z skupino, mora navesti svoje ime, kar ga nedvoumno identificira znotraj skupine. V zgoraj navedeni ukazni vrstici je še en zelo pomemben detajl — posebna oznaka „>“. Ta posebna oznaka filtrira sporočila in pušča le tista, ki še nikoli niso bila dostavljena.

Prav tako lahko v posebnih primerih navedete dejansko oznako, kot je 0 ali katera koli druga veljavna oznaka. V tem primeru ukaz XREADGROUP vam bo vrnil zgodovino sporočil s statusom „pending“, ki so bila dostavljena določenemu naročniku (Alice), a še niso bila potrjena z ukazom XACK.

To obnašanje lahko preverimo tako, da takoj navedemo oznako 0, brez možnosti COUNT. Videli bomo samo eno čakajoče sporočilo, torej sporočilo z jabolkom:

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "sporočilo"
            2) "jabolko"

Vendar, če potrdimo sporočilo kot uspešno obdelano, se ne bo več prikazovalo:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (prazna lista ali zbirka)

Zdaj je čas, da Bob kaj prebere:

> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569498055-0
         2) 1) "sporočilo"
            2) "oranžna"
      2) 1) 1526569506935-0
         2) 1) "sporočilo"
            2) "jagoda"

Bob, član skupine mygroup, je prosil za največ dve sporočili. Ukaz poroča le o nedostavljenih sporočilih zaradi posebne oznake „>“. Kot vidite, sporočilo „jabolko“ ne prikazuje, saj je bilo že dostavljeno Alica, zato Bob prejme „oranžna“ in „jagoda“.

Tako lahko Alice, Bob in kateri koli drug naročnik skupine berejo različna sporočila iz istega toka. Prav tako lahko berejo svojo zgodovino neobdelanih sporočil ali označijo sporočila kot obdelana.

Obstaja nekaj stvari, na katere je treba paziti:

Ko naročnik prebere sporočilo z ukazom XREADGROUP, to sporočilo preide v stanje „pending“ in postane dodeljeno temu določenemu naročniku. Drugi naročniki skupine ne bodo mogli prebrati tega sporočila.
Naročniki se samodejno ustvarijo ob prvem omembi, ni potrebno, da jih izrecno ustvarite.
S pomočjo XREADGROUP Lahko hkrati berete sporočila iz različnih tokov, vendar morate najprej ustvariti skupine z istim imenom za vsak tok s pomočjo XGROUP

Obnova po zrušitvi

Potrošnik se lahko po napaki obnovi in ponovno prebere svoj seznam sporočil s statusom „pending“. Vendar pa v resničnem svetu potrošniki lahko končno odpovejo. Kaj se zgodi z visočimi sporočili potrošnika, če se po napaki ni uspel obnoviti?
Consumer Group ponuja funkcijo, ki se uporablja natančno za taka slučaja — ko je potrebno zamenjati lastnika sporočil.

Najprej je treba poklicati ukaz XPENDING, ki prikaže vsa sporočila skupine s statusom „pending“. V svoji najpreprostejši obliki se ukaz pokliče z dvema argumentoma: imenom toka in imenom skupine:

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

Ukaz je izpisal število neobdelanih sporočil za celo skupino in za vsakega potrošnika. Imamo samo Boba z dvema neobdelanima sporočiloma, ker je edino sporočilo, ki ga je zahtevala Alice, bilo potrjeno s pomočjo XACK.

Lahko zahtevamo dodatne informacije z uporabo več argumentov:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — obseg identifikatorjev (lahko uporabite „-“ in „+“)
{count} — število poskusov dostave
{consumer-name} — ime skupine

> 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

Zdaj imamo podrobnosti za vsako sporočilo: identifikator, ime potrošnika, čas zastoja v milisekundah in, končno, število poskusov dostave. Imamo dva sporočila od Boba, ki sta v zastoju že 74170458 milisekund, kar je približno 20 ur.

Upoštevajte, da nam nihče ne preprečuje, da preverimo, kakšna je bila vsebina sporočila, preprosto z uporabo XRANGE.

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

Preprosto moramo ponoviti isti identifikator dvakrat v argumentih. Zdaj, ko imamo neko predstavo, lahko Alice sklene, da Bob verjetno ne bo obnovljen po 20 urah zastoja, in da je čas, da zahteva ta sporočila in ponovno zažene obdelavo namesto Boba. Za to uporabimo ukaz XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Mit diesem Befehl können wir eine "fremde" Nachricht abrufen, die noch nicht verarbeitet wurde, indem wir den Besitzer auf {consumer} ändern. Wir können jedoch auch eine minimale Leerlaufzeit von {min-idle-time} anbieten. Dies hilft, Situationen zu vermeiden, in denen zwei Kunden gleichzeitig versuchen, den Besitzer derselben Nachrichten zu wechseln.

Klient 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Klient 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Der erste Kunde setzt die Leerlaufzeit zurück und erhöht den Zustellzähler. Daher kann der zweite Kunde ihn nicht anfordern.

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

Die Nachricht wurde erfolgreich von Alice angefordert, die nun die Nachricht verarbeiten und bestätigen kann.

Aus dem obigen Beispiel ist ersichtlich, dass die erfolgreiche Ausführung der Anfrage den Inhalt der Nachricht selbst zurückgibt. Dies ist jedoch nicht zwingend erforderlich. Die Option JUSTID kann verwendet werden, um nur die Identifikatoren der Nachricht zurückzugeben. Dies ist nützlich, wenn Sie sich nicht für die Details der Nachricht interessieren und die Leistung des Systems steigern möchten.

Zustellzähler

Der Zähler, den Sie in der Ausgabe beobachten XPENDING ist die Anzahl der Zustellungen jeder Nachricht. Dieser Zähler erhöht sich auf zwei Arten: wenn die Nachricht erfolgreich angefordert wird über XCLAIM oder wenn der Aufruf verwendet wird XREADGROUP.

Es ist normal, dass einige Nachrichten mehrmals zugestellt werden. Das Wichtigste ist, dass am Ende alle Nachrichten verarbeitet wurden. Manchmal treten bei der Verarbeitung einer Nachricht Probleme auf, sei es aufgrund einer Beschädigung der Nachricht selbst oder weil die Verarbeitung der Nachricht einen Fehler im Handler-Code verursacht. In diesem Fall kann es vorkommen, dass diese Nachricht niemand verarbeiten kann. Da wir einen Zustellversuchszähler haben, können wir diesen verwenden, um solche Situationen zu entdecken. Daher ist es sinnvoll, sobald der Zustellzähler die von Ihnen festgelegte hohe Zahl erreicht, diese Nachricht in einen anderen Stream zu verschieben und den Systemadministrator zu benachrichtigen.

Status der Streams

Komanda XINFO wird verwendet, um verschiedene Informationen über den Stream und seine Gruppen abzurufen. Zum Beispiel sieht der grundlegende Befehl wie folgt aus:

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

Zgornja ukaz prikazuje skupne informacije o navedenem toku. Zdaj nekoliko bolj zapleten primer:

> XINFO GROUPS mystream
1) 1) ime
   2) "mygroup"
   3) porabniki
   4) (celo število) 2
   5) čakajoči
   6) (celo število) 2
2) 1) ime
   2) "some-other-group"
   3) porabniki
   4) (celo število) 1
   5) čakajoči
   6) (celo število) 0

Zgornji ukaz prikazuje skupne informacije o vseh skupinah navedenega toka

> XINFO CONSUMERS mystream mygroup
1) 1) ime
   2) "Alice"
   3) čakajoči
   4) (celo število) 1
   5) brez dela
   6) (celo število) 9104628
2) 1) ime
   2) "Bob"
   3) čakajoči
   4) (celo število) 1
   5) brez dela
   6) (celo število) 83841983

Zgornji ukaz prikazuje informacije o vseh naročnikih navedenega toka in skupine.
Če pozabite sintakso ukaza, preprosto poiščite pomoč pri samem ukazu:

> XINFO HELP
1) XINFO {podukaz} arg arg ... arg. Podukazi so:
2) CONSUMERS {ključ} {ime_skupine}  -- Prikaži porabniške skupine skupine {ime_skupine}.
3) GROUPS {ključ}                 -- Prikaži skupine porabnikov toka.
4) STREAM {ključ}                 -- Prikaži informacije o toku.
5) HELP                         -- Natisnite to pomoč.

Omejitev velikosti toka

Mnoga številna aplikacija ne želi zbirati podatkov v toku večno. Pogosto je koristno imeti največje dovoljeno število sporočil v toku. V drugih primerih je koristno prenesti vsa sporočila iz toka v drugo trajno skladišče, ko dosežemo določeno velikost toka. Velikost toka lahko omejimo s параметром MAXLEN v ukazu 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
(celo število) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

Pri uporabi MAXLEN se stare zapisane informacije samodejno izbrišejo, ko dosežemo določeno dolžino, zato ima tok stalno velikost. Vendar pa je obrezovanje v tem primeru izvedeno na nekoliko manj učinkovit način v pomnilniku Redis. Situacijo je mogoče izboljšati na naslednji način:

XADD mystream MAXLEN ~ 1000 * ... vnosni polji tukaj ...
Argument ~ v zgornjem primeru pomeni, da nam ni nujno treba omejevati dolžine toka na specifično vrednost. V našem primeru je lahko katero koli število večje ali enako 1000 (na primer, 1000, 1010 ali 1030). Preprosto smo jasno navedli, da želimo, da naš tok hrani najmanj 1000 zapisov. To naredi delo s pomnilnikom veliko učinkovitejše znotraj Redis.

Obstaja tudi ločen ukaz XTRIM, ki izvaja isto:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Trajna shranitev in replikacija

Redis Stream se asinhrono replicira na slave nodove i čuva se u AOF (snapshot svih podataka) i RDB (log svih operacija pisanja). Takođe se podržava replikacija stanja Consumer Groups. Dakle, ako je poruka u statusu „pending“ na master nodu, na slave nodovima će imati isti status.

Brisanje pojedinačnih elemenata iz toka

Za brisanje poruka postoji posebna komanda XDEL. Komanda prima ime toka, nakon kojeg slede identifikatori poruka koje je potrebno obrisati:

> 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šćenju ove komande treba imati na umu da se zapravo memorija neće osloboditi odmah.

Toki nulte dužine

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 nuspojava, sama struktura podataka biće uklonjena iz memorije. Na primer, sortirani skup će biti potpuno uklonjen kada poziv ZREM obriše poslednji element. Umesto toga, tokovi mogu ostati u memoriji, čak i kada nemaju nijedan element unutar sebe.