Redis Stream — uus abstraktne andmetüüp, mis tutvustati Redis 5.0 versiooniga.
Konceptsiooniliselt on Redis Stream nagu List, kuhu saate lisada kirjeid. Igal kirjel on ainulaadne identifikaator. Vaikimisi genereeritakse identifikaator automaatselt ja sisaldab ajatemplit. Seetõttu saate pärida ajavahemikke kirjed või saada uusi andmeid voogusid, nagu Unix käsk „tail -f“ loeb logifaili ja ootab uusi andmeid. Pange tähele, et voogu saavad samaaegselt kuulata mitu klienti, nagu paljud „tail -f“ protsessid saavad samal ajal faili lugeda, konflikte tekitamata.

Uue andmetüübi kõiki eeliseid mõistmiseks meenutame kiiresti varasemaid Redis-struktuure, mis osaliselt kordavad Redis Stream'i funktsionaalsust.

Redis PUB/SUB

Redis Pub/Sub on lihtne sõnumite süsteem, mis on juba teie võtme-väärtuse salvestusse sisse ehitatud. Kuid lihtsuse eest tuleb maksta:

Kui väljund (publisher) mingil põhjusel välja langeb, kaotab ta kõik oma tellijad.
Väljundil peab olema täpne aadress kõigist oma tellijatest.
Väljund võib koormata oma tellijaid, kui andmeid avaldatakse kiiremini, kui neid töödeldakse.
Sõnum eemaldatakse väljundi puhvrist kohe pärast avaldamist, sõltumata sellest, kui paljudele tellijatele see on edastatud ja kui kiiresti nad suudavad seda sõnumit töödelda.
Kõik tellijad saavad sõnumi samal ajal. Tellijad peavad omavahel leppima kokku sama sõnumi töötlemise järjekorras.
Puudub sisseehitatud mehhanism sõnumi eduka töötlemise kinnitamiseks tellija poolt. Kui tellija saab sõnumi ja tegevuse käigus langeb välja, ei saa väljund sellest teada.

Redis List

Redis List on andmestruktuur, mis toetab blokeerivaid lugemisekäsku. Saate lisada ja lugeda sõnumeid nimekirja algusest või lõpust. Selle struktuuri põhjal on võimalik luua korralik virn või järjekord teie jaotatud süsteemis, ja enamasti on seda piisavalt. Peamised erinevused Redis Pub/Sub'i omadest:

Sõnum edastatakse ühele kliendile. Esimene blokeeritud lugemise klient saab andmed esimesena.
Clint peab ise algatama iga sõnumi lugemise operatsiooni. List ei tea midagi klientidest.
Sõnumid säilivad seni, kuni keegi need loeb või selgelt kustutab. Kui olete seadistanud Redis serveri niimoodi, et see salvestab andmed kettale, siis süsteemi usaldusväärsus paraneb oluliselt.

Sissejuhatus Streami

Kirje lisamine voogu

Meeskond XADD lisab voogu uue kirje. Kirje ei ole lihtsalt rida, vaid see koosneb ühest või mitmest võtme-väärtuse paarist. Seega on iga kirje juba struktureeritud ja meenutab CSV faili struktuuri.

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

Ülaltoodud näites lisame voogu nimega (võtmena) «mystream» kaks väljast: «sensor-id» ja «temperature» väärtustega «1234» ja «19.8» vastavalt. Teise argumendina võtab käsk identifikaatori, mis antakse kirjele – see identifikaator tuvastab iga kirje voos ühemõtteliselt. Kuid sel korral edastasime *, kuna soovime, et Redis genereeriks meile uue identifikaatori. Iga uus identifikaator suureneb. Seega on iga uus kirje eelnevate kirjete suhtes suurema identifikaatoriga.

Identifikaatori formaat

Kirje identifikaator, mille käsk XADD, koosneb kahest osast:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix aeg millisekundites (aeg serverilt Redis). Kui aga praegune aeg on sama või väiksem kui eelmise kirje aeg, kasutatakse eelmise kirje ajatemplit. Seega, kui serveri aeg tagastab minevikku, säilitab uus identifikaator ikkagi suurenemise omaduse.

sequenceNumber kasutatakse samal millisekundil loodud kirje jaoks. sequenceNumber suurendatakse 1 võrra eelneva kirje suhtes. Kuna sequenceNumber on 64 bitti pikk, ei tohiks tegelikult tekkida piirangut, kui palju kirjeid saab genereerida ühe millisekundi jooksul.

Nende identifikaatorite formaat võib esmapilgul tunduda kummaline. Kahtlev lugeja võib küsida, miks on aeg osa identifikaatorist. Põhjuseks on see, et Redis-puud toetavad identifikaatorite alusel vahemikku päringute tegemist. Kuna identifikaator on seotud salvestise loomise ajaga, siis võimaldab see küsida aegu vahemikus. Vaadakem konkreetset näidet, kui liigume käsu uurimise juurde. XRANGE.

Kui kasutajal on mingil põhjusel vaja määrata oma identifikaator, mis nt. on seotud mingi välist süsteemiga, siis võime selle edastada käsule XADD tähe * asemel nagu allpool näidatud:

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

Pange tähele, et sel juhul peate ise jälgima identifikaatori suurenemist. Meie näites on minimaalne identifikaator "0-1", seega käsk ei võta vastu veel ühte identifikaatorit, mis on võrdne või väiksem kui "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

Kirjete arv voos

Voos olevate kirjete arvu saab lihtsalt saada käsuga XLEN. Selle näite puhul tagastab see käsk järgmise väärtuse:

> XLEN somestream
(integer) 2

Vahemiku päringud — XRANGE ja XREVRANGE

Andmete pärimiseks vahemiku alusel peame määrama kaks identifikaatorit — vahemiku alguse ja lõpu. Tagastatav vahemik sisaldab kõiki elemente, sealhulgas piire. Samuti on olemas kaks erilise identifikaatorit „-” ja „+”, mis vastavad vastavalt kõige väiksemale (esimene salvestis) ja kõige suuremale (viimane salvestis) identifikaatorile voos. Allolev näide väljastab kõik voogu kuuluvad salvestised.

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

Iga tagastatud salvestis esindab kaht elementi: identifikaatori ja võtme-väärtuse paari loendi. Oleme juba maininud, et salvestise identifikaatorid on ajaga seotud. Seetõttu saame küsida teatud ajavahemiku vahemikku. Siiski saame päringus määrata mitte täieliku identifikaatori, vaid ainult Unix aja, jättes välja osa, mis on seotud sequenceNumberVahele jäetud identifikaatori osa omistatakse automaatselt vahemiku alguses nullile ja vahemiku lõpus maksimaalsele võimalikule väärtusele. Allpool on näide, kuidas küsida vahemikku, mis on kaks millisekundit.

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

Meil on selles vahemikus ainult üks kande, kuid reaalses andmehulkades võib tagastatav tulemus olla tohutu. Seetõttu XRANGE toetab valikut COUNT. Määrates arvu, saame lihtsalt saada esimesed N kannet. Kui me peame saama järgmised N kannet (leheküljestamine), võime kasutada viimati saadud identifikaatorit, suurendada seda sequenceNumber ühe võrra ja küsida uuesti. Vaatame seda järgmises näites. Alustame 10 elemendi lisamisega läbi XADD (eeldame, et müügivoog mystream on juba täidetud 10 elemendiga). Itereeringu alustamiseks, saades iga käsu kohta 2 elementi, alustame täielikust vahemikust, kuid COUNT on seatud 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"

Jätkamiseks järgmise kahe elemendi iteratsioonis peame valima viimase saadud identifikaatori, st 1519073279157-0, ja lisama 1 sequenceNumber.
saadud identifikaatorile, Antud juhul 1519073279157-1, saab nüüd kasutada uue argumendina vahemiku alguseks järgmises kõnes 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"

Ja nii edasi. Kuna keerukus XRANGE on O(log (N)) otsimiseks ja siis O(M) M elemendi tagastamiseks, on iga iteratsiooni samm kiire. Seega saab XRANGE efektiivselt vooge itererida.

Meeskond XREVRANGE on ekvivalent XRANGE, kuid tagastab elemendid vastupidises järjekorras:

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

Pange tähele, et käsk XREVRANGE võtab argumentide osas vahemiku start ja stop vastupidises järjekorras.

Uute kannete lugemine XREAD abil

Sageli on vaja tellida voogu ja saada vaid uusi sõnumeid. See kontseptsioon võib tunduda sarnane Redis Pub/Sub-iga või blokeeriva Redis Loendiga, kuid Redis Streami kasutamisel on põhimõttelised erinevused:

Iga uus sõnum jagatakse vaikimisi igale tellijale. See käitumine erineb blokeerivast Redis List'ist, kus uus sõnum loetakse ainult ühe tellija poolt.
Samas kui Redis Pub/Sub kõik sõnumid unustatakse ja kunagi ei salvestata, salvestatakse Stream'is kõik sõnumid määramata ajaks (kui klient selgesõnaliselt kustutamist ei taotle).
Redis Stream võimaldab piirata juurdepääsu sõnumitele ühe voolu sees. Konkreetne tellija näeb ainult oma isiklikku sõnumite ajalugu.

Saate voole tellida ja uusi sõnumeid vastu võtta, kasutades käsku XREAD. See on natuke keerulisem kui XRANGE, seetõttu alustame esmalt lihtsamate näidetega.

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

Ülaltoodud näites on määratud mitteblokkeeriv vorm XREAD. Pange tähele, et valik COUNT ei ole kohustuslik. Tegelikult on ainsa kohustusliku valikuna käskluste seas STREAMS, mis määrab voogude nimekirja koos vastava maksimaalse tuvastajaga. Me kirjutasime „STREAMS mystream 0“ — me tahame saada kõik kirjed voost mystream, mille tuvastaja on suurem kui „0-0“. Nagu näha näites, tagastab käsk voolu nime, kuna saame tellida mitu voolu korraga. Me oleksime võinud kirjutada näiteks „STREAMS mystream otherstream 0 0“. Pange tähele, et pärast valikut STREAMS peame esmalt esitama kõikide vajalike voogude nimed ja alles seejärel tuvastajate nimekirja.

Selles lihtsas vormis ei tee käsk midagi erilist võrreldes XRANGE. Kuid huvitav on see, et saame hõlpsasti muuta XREAD blokeerivaks käskuks, määrates argumendiks BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
Ülaltoodud näites on määratud uus BLOCK valik, mille ooteaeg on 0 millisekundit (see tähendab lõpmatult oodata). Veelgi enam, selle asemel et edastada tavaline tuvastaja voole mystream, edastati eriline tuvastaja $. See eriline tuvastaja tähendab, et XREAD tuleb kasutada tuvastajana maksimaalse tuvastajana voos mystream. Nii et me saame uusi sõnumeid alates hetkest, mil alustasime kuulamist. Teatud mõttes on see sarnane Unix'i käsule „tail -f“.

Pöörake tähelepanu, et BLOCK valiku kasutamisel ei pruugi meil olla vaja kasutada erilist identifikaatorit $. Me saame kasutada iga olemasolevat identifikaatorit voos. Kui meeskond suudab meie taotlust kohe teenindada, ilma lukustamata, siis nad teevad seda; vastasel juhul nad lukustuvad.

Lukustav XREAD võib samuti jälgida mitu voogu korraga, tuleb lihtsalt näidata nende nimesid. Sel juhul tagastab meeskond esimese voolu salvestuse, kuhu andmed tulid. Esimene tellija, kes on antud voos lukustatud, saab andmed kõige varem.

Tellijate grupid

Mõnedes ülesannetes tahame piiritleda tellijate juurdepääsu sõnumitele ühes voos. Näide, kui see võib olla kasulik — sõnumite järjekord töötajatega, kes saavad erinevaid voosõnumeid, võimaldades sõnumite töötlemise skaleerimist.

Kui kujutame ette, et meil on kolm tellijat C1, C2, C3 ja voog, mis sisaldab sõnumeid 1, 2, 3, 4, 5, 6, 7, siis sõnumite teenindamine toimub nagu alloleval diagrammil:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Selle efekti saavutamiseks kasutab Redis Stream kontseptsiooni, mida nimetatakse tellijate grupiks. See kontseptsioon sarnaneb pseudotellijaga, mis saab andmeid voost, kuid tegelikult teenindatakse seda mitme tellijaga grupi sees, pakkudes teatud garanteeringuid:

Iga sõnum edastatakse erinevatele tellijatele grupi sees.
Grupi siseselt tunnistatakse tellijad nime järgi, mis on suur- ja väiketähti arvesse võtav string. Kui mõni tellija ajutiselt grupist välja jääb, siis võib ta taastuda grupis oma unikaalse nime abil.
Iga tellijate grupp järgib kontseptsiooni "esimene lugemata sõnum". Kui tellija küsib uusi sõnumeid, võib ta saada ainult neid sõnumeid, mida pole varem ühelegi tellijale grupis edastatud.
On olemas käsk, millega tellija kinnitab sõnumi eduka töötlemise. Kuni seda käsku ei ole kutsutud, jääb küsitud sõnum staatuse „oote“ alla.
Tellijate grupi sees võib iga tellija küsida sõnumite ajalugu, mis on talle edastatud, kuid pole veel töödeldud (staatuses „oote“).

Teatud mõttes võib grupi seisundit esitada järgmiselt:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" ootel sõnumitega  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" ootel sõnumitega 
|       ... (ja nii edasi)                             
+----------------------------------------+

Nüüd on aeg tutvuda peamiste käskudega Consumer Group'i jaoks, nimelt:

XGROUP kasutatakse gruppide loomiseks, hävitamiseks ja haldamiseks
XREADGROUP kasutatakse voogude lugemiseks grupi kaudu
XACK — see käsk võimaldab tellijal märkida sõnumi edukalt töödelduna

Consumer Group'i loomine

Oletame, et voog mystream juba eksisteerib. Siis näeb grupi loomise käsk välja järgmiselt:

> XGROUP CREATE mystream mygroup $ OK
Grupi loomisel peame edastama ID, millest alates grupp hakkab sõnumeid saama. Kui soovime lihtsalt saada kõiki uusi sõnumeid, võime kasutada erilist identifikaatorit $ (nagu meie eespool olevas näites). Kui selle asemel kasutatakse erilise identifikaatori asemel 0, on grupile saadaval kõik voogude sõnumid.

Nüüd, kui grupp on loodud, saame kohe hakata sõnumeid lugema kasutades käsku XREADGROUP. See käsk on väga sarnane XREAD ja toetab valikulist valikut BLOCK. Kuid alati tuleb kasutada kohustuslikku valikut GROUP, mis peab olema alati määratud kahe argumendiga: grupi nimi ja tellija nimi. Valikut COUNT toetatakse samuti.

Enne voogude lugemist paneme sinna mõned sõnumid:

> 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

Ja nüüd proovime seda voogu lugeda grupi kaudu:

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

Ülaltoodud käsk ütleb sõna-sõnalt järgmist:

«Mina, Alice-tellija, mygroup'i liikmena, tahan lugeda voost mystream ühte sõnumit, mida ei ole kunagi kellelegi varem edastatud.»

Iga kord, kui tellija teostab operatsiooni grupiga, peab ta märkima oma nime, tuvastades ennast grupis. Ülaltoodud käsus on veel üks väga oluline detail - eriline identifikaator „>”. See eriline identifikaator filtreerib sõnumid, jättes alles need, mis pole kunagi varem kohale toimetatud.

Samuti võite erandlikel juhtudel märkida reaalse identifikaatori, näiteks 0 või mõne muu kehtiva identifikaatori. Sel juhul käsk XREADGROUP tagastab teile sõnumite ajalugu staatusega „ootele”, mis on toimetatud antud tellijale (Alice), kuid pole veel kinnitatud käsuga XACK.

Saame seda käitumist kontrollida, märkides kohe identifikaatori 0, ilma valikuta LOEND. Näeme lihtsalt ühte ainsat ootel sõnumit, nimelt sõnumit õunast:

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "sõnum"
            2) "õun"

Kuid kui kinnitame sõnumi edukalt töödelduna, ei kuvata seda enam:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (tühi loend või kogum)

Nüüd on Bobil aeg midagi lugeda:

> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569498055-0
         2) 1) "sõnum"
            2) "apelsin"
      2) 1) 1526569506935-0
         2) 1) "sõnum"
            2) "maasikas"

Bob, mygroup'i liige, küsis mitte rohkem kui kahte sõnumit. Käsk raporteerib vaid kohale toimetamata sõnumeid erilise identifikaatori „>” tõttu. Nagu näete, ei kuvata sõnumit „õun”, kuna see on juba Alicele toimetatud, seetõttu saab Bob „apelsini” ja „maasika”.

Nii saavad Alice, Bob ja iga teine grupi tellija lugeda erinevaid sõnumeid ühest ja samast voogust. Samuti saavad nad lugeda oma töödeldud sõnumite ajalugu või märkida sõnumeid töödelduna.

On mõned asjad, mida tasub meeles pidada:

Kui tellija loeb sõnumit käsuga XREADGROUP, liigub see sõnum olekusse „ootele” ja kinnitatakse selle konkreetse tellija alla. Teised grupi tellijad ei saa seda sõnumit lugeda.
Tellijad luuakse automaatselt esmakordsel mainimisel, ei ole vaja neid eraldi luua.
EL-i abil XREADGROUP Saate lugeda teateid mitmest erinevast voost samaaegselt, kuid selleks peab teil eelnevalt olema loodud rühmad, millel on iga voolu jaoks sama nimi, kasutades XGROUP

Tõrke taastamine

Tellija võib taastuda tõrkest ja lugeda oma sõnumite loendit, mille olek on "pending". Kuid tegelikus maailmas võivad tellijad jääda püsivalt ebaõnnestunuks. Mis juhtub tellija ooterežiimis olevate sõnumitega, kui ta ei suuda tõrget taastada?
Kasutajagrupi funktsioon pakub lahendust selliste juhtumite jaoks — kui on vajalik sõnumite omandi õiguse vahetamine.

Esiteks tuleb käivitada käsk XPENDING, mis kuvab kõik rühma sõnumid olekuga "pending". Oma kõige lihtsamas vormis käivitatakse käsk vaid kahe argumendiga: voolu nimi ja rühma nimi:

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

Käsk trükib välja töötlemata sõnumite arvu kogu grupi ja iga tellija jaoks. Meil on ainult Bob, kellel on kaks töötlemata sõnumit, kuna Alice'i taotletud ainus sõnum kinnitati läbi XACK.

Saame küsida lisainfot, kasutades rohkem argumente:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — identifikaatorite vahemik (võib kasutada "-" ja "+")
{count} — katsete arv
{consumer-name} — rühma nimi

> 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

Nüüd on meil iga sõnumi kohta üksikasjad: identifikaator, tellija nimi, ooteaeg millisekundites ning lõpuks katsete arv. Meil on kaks sõnumit Bobilt, mis ootavad 74170458 millisekundit, umbes 20 tundi.

Pange tähele, et ei miski takista meil kontrollida, milline oli sõnumite sisu, kasutades lihtsalt XRANGE.

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

Peame lihtsalt kordama sama identifikaatorit kaks korda argumentides. Nüüd, kui meil on mingi ettekujutus, võib Alice järeldada, et pärast 20 tunni ootamist ei taastu Bob tõenäoliselt ning on aeg need sõnumid taotleda ja nende töötlemisega jätkata Bobi asemel. Selleks kasutame käsku XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Selle käsu abil saame hankida «võõra» sõnumi, mis pole veel töödeldud, muutes omaniku {consumer} -iks. Samuti saame pakkuda minimaalset seismisaega {min-idle-time}. See aitab vältida olukorda, kus kaks klienti üritavad samal ajal sama sõnumi omaniku vahetada:

Kliendi 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Kliendi 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Esimene klient nullib seismisaja ja suurendab kohaletoimetamise arvu loendurit. Seega ei saa teine klient seda nõuda.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "sõnum"
      2) "oranž"

Sõnum on edukalt nõudnud Alice, kes nüüd saab sõnumit töödelda ja kinnitada.

Ülalkirjeldatud näitest on näha, et eduka päringu täitmine tagastab sõnumi sisu. Kuid see pole kohustuslik. Opetsioon JUSTID saab kasutada ainult sõnumi id-de tagastamiseks. See on kasulik, kui teid ei huvita sõnumi detailid ja soovite süsteemi jõudlust suurendada.

Kohaletoimetamise loendur

Loendur, mida näete väljundis XPENDING — on iga sõnumi kohaletoimetamiste arv. See loendur suureneb kahel viisil: kui sõnum on edukalt taotletud läbi XCLAIM või kui kasutatakse kutset XREADGROUP.

On normaalne, et mõned sõnumid toimetatakse mitu korda. Peamine on see, et kõik sõnumid oleksid lõpuks töödeldud. Mõnikord tekivad probleeme sõnumi töötlemisel, mis tulenevad sõnumi enda kahjustumisest või sõnumi töötlemine tekitab töötleja koodi vea. Sellisel juhul võib juhtuda, et seda sõnumit ei saa keegi töödelda. Kuna meil on kohaletoimetamise katsete loendur, saame seda loendurit kasutada selliste olukordade tuvastamiseks. Seetõttu, kui kohaletoimetamiste loendur saavutab teie määratud suure numbri, on tõenäoliselt mõistlik selline sõnum viia teise voogude ja saata süsteemiadministraatorile teade.

Voolude olek

Meeskond XINFO kasutatakse voolu ja selle gruppide erineva teabe pärimiseks. Näiteks näeb käsu põhivälimus järgmine välja:

> 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) "sõnum"
       2) "banaan"

Ülemine käsk näitab määratud voogu üldteavet. Nüüd veidi keerulisem näide:

> XINFO GROUPS mystream
1) 1) nimi
   2) "mygroup"
   3) tarbijad
   4) (täisarv) 2
   5) ootel
   6) (täisarv) 2
2) 1) nimi
   2) "some-other-group"
   3) tarbijad
   4) (täisarv) 1
   5) ootel
   6) (täisarv) 0

Ülemine käsk näitab kõigi määratud voogu grupi üldteavet.

> XINFO CONSUMERS mystream mygroup
1) 1) nimi
   2) "Alice"
   3) ootel
   4) (täisarv) 1
   5) mitteaktiivne
   6) (täisarv) 9104628
2) 1) nimi
   2) "Bob"
   3) ootel
   4) (täisarv) 1
   5) mitteaktiivne
   6) (täisarv) 83841983

Ülemine käsk näitab teavet kõigi määratud voogu ja grupi tellijate kohta.
Kui unustate käskluse süntaksi, puudutage lihtsalt abi saamiseks käsku ise:

> XINFO HELP
1) XINFO {alampäring} arg arg ... arg. Alamkäsklused on:
2) CONSUMERS {võti} {grupi nimi}  -- Näita grupi {grupi nimi} tarbijagrupe.
3) GROUPS {võti}                 -- Näita voogu tarbijagrupe.
4) STREAM {võti}                 -- Näita teavet voo kohta.
5) HELP                         -- Prindi see abi.

Voo suuruse piirang

Paljud rakendused ei soovi koguda andmeid voogu igavesti. Sageli on kasulik omada maksimaalset lubatud sõnumite arvu voos. Teistes olukordades on kasulik edastada kõik sõnumid voost teise püsivasse salvestusse, kui sätestatud voosuurus on saavutatud. Voo suurust saab piirata MAXLEN parameetriga käsus. XADD:

> XADD mystream MAXLEN 2 * väärtus 1
1526654998691-0
> XADD mystream MAXLEN 2 * väärtus 2
1526654999635-0
> XADD mystream MAXLEN 2 * väärtus 3
1526655000369-0
> XLEN mystream
(täisarv) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "väärtus"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "väärtus"
      2) "3"

MAXLENi kasutamisel kustutatakse vanad kirjed automaatselt, kui määratud pikkus on saavutatud, seega on voog alaliselt suurusega. Kuid lõikamine ei toimu kõige tõhusamal viisil Redis mälus. Seda olukorda saab parandada järgmiselt:

XADD mystream MAXLEN ~ 1000 * ... kirjeväljad siia ...
Argument ~ ülaltoodud näites tähendab, et meil ei ole tingimata vaja voogu piirata kindla väärtusega. Meie näites võib see olla mis tahes number, mis on suurem või võrdne 1000 (näiteks 1000, 1010 või 1030). Lihtsalt oleme selgelt öelnud, et soovime, et meie voog salvestaks vähemalt 1000 kirjet. See muudab mälu haldamise palju tõhusamaks Redis'is.

Olemas on ka eraldi käsk XTRIM, mis täidab sama ülesannet:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Püsiv salvestamine ja replikatsioon

Redis Stream asünkroonselt replikeeritakse slave sõlmedele ning salvestatakse AOF (kõigi andmete momentne seis) ja RDB (kõikide kirjutamisoperatsioonide logi) failidesse. Samuti toetatakse Consumer Groups'i oleku replikeerimist. Seega, kui sõnum on master sõlmes staatuses „pending“, siis slave sõlmedes on sellel sõnumil sama staatus.

Üksikute elementide eemaldamine voost

Sõnumite eemaldamiseks on olemas spetsiaalne käsk XDEL. Käsk võtab vastu voolu nime, millele järgneb sõnumite identifikaatorid, mille tuleb eemaldada:

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

Seda käsku kasutades tuleb arvestada, et mälu vabastamine ei toimu kohe.

Nullpikkusega vood

Erinevus voogude ja muude Redis'i andmestruktuuride vahel seisneb selles, et kui teised andmestruktuurid ei sisalda enam elemente, eemaldatakse andmestruktuur ise mälust kõrvaltoimena. Näiteks, järjestatud kogum eemaldatakse täielikult, kui ZREM käsk eemaldab viimase elemendi. Vastupidi, vood võivad jääda mällu, isegi kui need ei sisalda ühtegi elementi.

Kokkuvõte

Redis Stream sobib ideaalselt sõnumitevahetuse vahendite, sõnumijärjekordade, ühtsete ajakirjade ja vestlussüsteemide, mis salvestavad ajalugu, loomiseks.

Kuidas ütles kunagi Niklaus Wirth, programmid — see on algoritmid pluss andmestruktuurid, ja Redis annab teile juba mõlemat.

Allikas: habr.com