A Redis Stream — egy új absztrakt adattípus, amely a Redis 5.0-ás verziójával került bemutatásra.
Konceptuálisan a Redis Stream egy lista, amelybe rekordokat adhat hozzá. Minden rekordnak van egyedi azonosítója. Alapértelmezés szerint az azonosítót automatikusan generálják, és időbélyeget tartalmaz. Ezért időszakok szerint kérheti le a rekordok tartományát, vagy kaphat új adatokat, ahogy az a streambe érkezik, hasonlóan a Unix parancshoz, a «tail -f»-hez, amely naplófájlt olvas és várakozik az új adatokra. Vegye figyelembe, hogy a streamet egyszerre több kliens is figyelheti, akárcsak a sok «tail -f» folyamat, amelyek egyszerre olvashatják a fájlt, anélkül, hogy összeütköznének egymással.

Annak érdekében, hogy megértsük az új adattípus minden előnyét, gyorsan emlékezzünk vissza a Redis már régóta létező struktúráira, amelyek részben megismétlik a Redis Stream funkcióit.

Redis PUB/SUB

A Redis Pub/Sub egy egyszerű üzenetküldő rendszer, amely már be van építve a kulcs-érték tárolójába. Azonban az egyszerűségért meg kell fizetni:

Ha a kiadó bármilyen okból leáll, elveszíti minden feliratkozóját.
A kiadónak tudnia kell az összes feliratkozójának pontos címét.
A kiadó túlterhelheti a feliratkozóit, ha az adatok gyorsabban publikálódnak, mint ahogy azok kezelik.
Az üzenet azonnal törlődik a kiadó puffere-ből a közzététel után, függetlenül attól, hogy hány feliratkozónak került kézbesítésre és hogy azok milyen gyorsan tudták feldolgozni az üzenetet.
Minden feliratkozó egyszerre kapja meg az üzenetet. A feliratkozóknak maguknak kellene valamilyen módon egyeztetniük az azonos üzenet feldolgozásának sorrendjét.
Nincs beépített mechanizmus a feliratkozó által az üzenet sikeres feldolgozásának megerősítésére. Ha a feliratkozó megkapta az üzenetet, de a feldolgozás közben leállt, a kiadó erről nem értesül.

Redis Lista

A Redis Lista egy olyan adatszerkezet, amely támogatja a blokkolt olvasási parancsokat. Üzeneteket adhat hozzá és olvashat el a lista elejéről vagy végéről. E struktúra alapján létrehozhat egy tisztességes verem- vagy sorstruktúrát az elosztott rendszeréhez, és ez a legtöbb esetben elegendő lesz. Az alapvető különbségek a Redis Pub/Sub-hoz képest:

Az üzenet egy klienshez jut el. Az első ügyfél, aki blokkolt olvasással hívja meg, először kapja az adatokat.
Clint-nek magának kell kezdeményeznie minden üzenet olvasását. A List nem tud semmit az ügyfelekről.
Az üzenetek addig maradnak tárolva, amíg valaki el nem olvassa őket, vagy kifejezetten el nem törli. Ha beállítottad a Redis szervert, hogy az adatokat a lemezre mentse, akkor a rendszer megbízhatósága jelentősen megnő.

Bevezetés a Streamhez

Bejegyzés hozzáadása a streamhez

A parancs XADD új bejegyzést ad a streamhez. A bejegyzés nem csupán egy szöveg, hanem egy vagy több kulcs-érték párból áll. Így tehát minden bejegyzés már struktúrált, és a CSV fájl struktúrájára emlékeztet.

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

A fenti példában a ‘mystream’ nevű (kulcsú) streamhez két mezőt adunk hozzá: ‘sensor-id’ és ‘temperature’ az ‘1234’ és ‘19.8’ értékekkel. A második argumentumként a bejegyzéshez rendelni kívánt azonosítót adunk meg — ez az azonosító egyértelműen azonosítja minden bejegyzést a streamben. Azonban ebben az esetben *, mert azt szeretnénk, ha a Redis generálna nekünk egy új azonosítót. Minden új azonosító növekedni fog. Így tehát minden új bejegyzés nagyobb azonosítót fog kapni az előző bejegyzésekkel összehasonlítva.

Azonosító formátuma

A parancs által visszaadott bejegyzés azonosítója, XADDkét részből áll:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix idő milliszekundumban (idő a szerver Redis). Ha azonban a jelenlegi idő megegyezik, vagy kisebb, mint a korábbi bejegyzés ideje, akkor a korábbi bejegyzés időbélyegét használja. Ezért, ha a szerver időpontja visszatér a múltba, az új azonosító még mindig megőrzi a növekvő tulajdonságát.

sequenceNumber amely a ugyanabban a milliszekundumban létrehozott bejegyzésekre vonatkozik. sequenceNumber 1-gyel növekszik az előző bejegyzéshez képest. Mivel sequenceNumber 64 bites mérettel rendelkezik, gyakorlatilag nem kell tartanod attól, hogy a létrehozható bejegyzések száma egy milliszekundum alatt korlátozott legyen.

Ezeknek az azonosítóknak a formátuma első ránézésre furcsának tűnhet. Egy bizalmatlan olvasó jogosan felteheti a kérdést, miért szerepel az idő az azonosítóban. Az ok az, hogy a Redis adatfolyamok támogatják az azonosítók alapján történő tartománykéréseket. Mivel az azonosító a rekord létrehozásának idejéhez kapcsolódik, ez lehetőséget ad az időtartományok lekérdezésére. Egy konkrét példát fogunk megnézni, amikor áttérünk a parancs tanulmányozására. XRANGE.

Ha bármilyen okból a felhasználónak szüksége van arra, hogy megadja a saját azonosítóját, amely például egy külső rendszerhez kapcsolódik, akkor azt átadhatjuk a parancsnak XADD a * jel helyett, ahogyan az alább látható:

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

Ne feledje, hogy ebben az esetben Önnek magának kell figyelnie az azonosító növelésére. Példánkban a minimális azonosító '0-1', ezért a parancs nem fogad el egy másik azonosítót, amely egyenlő vagy kisebb, mint '0-1'.

> XADD somestream 0-1 foo bar
(error) ERR Az XADD-ben megadott azonosító egyenlő vagy kisebb, mint a célstream legfelső eleme.

A rekordok száma az adatfolyamban

A stream rekordjainak számát egyszerűen lekérhetjük a XLENparancs használatával. Példánk számára ez a parancs a következő értéket adja vissza:

> XLEN somestream
(integer) 2

Tartománykérések — XRANGE és XREVRANGE

Ahhoz, hogy adatokat kérjünk le tartomány alapján, meg kell adnunk két azonosítót — a tartomány kezdete és vége. A visszaadott tartomány minden elemet tartalmaz, beleértve a határokat is. Két speciális azonosító is létezik, a '-' és a '+', amelyek a stream legkisebb (első rekord) és legnagyobb (utolsó rekord) azonosítóját jelentik. Az alábbi példa az adatfolyam összes rekordját megjeleníti.

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

Minden visszaadott rekord egy két elemből álló tömbot képvisel: az azonosítót és a kulcs-érték párok listáját. Már említettük, hogy a rekordok azonosítói az időhöz kapcsolódnak. Ezért lehetőségünk van egy adott időintervallum tartományának lekérdezésére. Azonban a kérésben nem szükséges a teljes azonosítót megadni, csak az Unix időt, az azt az azonosító részét kihagyva. sequenceNumberA hiányzó azonosító rész automatikusan 0-ra van állítva a tartomány elején, és a maximálisan lehetséges értékre a tartomány végén. Íme egy példa arra, hogyan kérhetünk egy két milliszekundumos tartományt.

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

Csak egy bejegyzésünk van ebben a tartományban, azonban a valós adathalmazon a visszaadott eredmények hatalmasak lehetnek. Ennek okán XRANGE támogatja a COUNT opciót. A megadott szám megadásával egyszerűen az első N bejegyzést kaphatjuk meg. Ha a következő N bejegyzést szeretnénk megkapni (oldalazás), az utolsó megkapott azonosítót felhasználhatjuk, és egyet hozzáadhatunk. sequenceNumber Egyet hozzáadunk, és újra kérhetjük. Nézzük meg ezt a következő példában. Kezdjük 10 elem hozzáadásával XADD (tegyük fel, hogy a mystream már 10 elemmel van feltöltve). Az iteráció elkezdéséhez, 2 elem nyerésével parancsonként, teljes tartományból indulunk, de COUNT értéke 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"

A következő két elem folytatásához ki kell választanunk az utolsó megkapott azonosítót, amely 1519073279157-0, és 1-et hozzáadni. sequenceNumber.
Az eredmény azonosító, ebben az esetben 1519073279157-1, most már új argumentumként használható a következő hívás kezdeti tartományaként. 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"

És így tovább. Mivel a bonyolultság XRANGE O(log (N))-ra ad vissza, majd O(M)-re az M elemet visszaadva, így minden iterációs lépés gyors. Tehát a XRANGE segítségével hatékonyan lehet iterálni az adatfolyamokat.

A parancs XREVRANGE egyenértékű XRANGE, de az elemeket fordított sorrendben adja vissza:

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

Note that the command XREVRANGE fordított sorrendben veszi figyelembe a start és stop tartomány argumentumokat.

Új bejegyzések olvasása az XREAD segítségével

Gyakran merül fel a feladat, hogy feliratkozzunk egy adatfolyamra, és csak az új üzeneteket kapjuk meg. Ez a koncepció hasonló lehet a Redis Pub/Sub-hoz vagy a blokkoló Redis Listához, de alapvető különbségek vannak a Redis Stream használatában:

Minden új üzenetet alapértelmezés szerint minden feliratkozó megkap. Ez a viselkedés eltér a blokkoló Redis List-től, ahol az új üzenetet csak egyetlen felhasználó olvassa el.
Míg a Redis Pub/Sub esetében minden üzenet elfelejtődik és soha nem kerül mentésre, a Stream összes üzenetet határozatlan ideig megőrzi (ha a kliens kifejezetten nem kérdőzi meg a törlést).
A Redis Stream lehetővé teszi az üzenetek hozzáférésének elkülönítését egyetlen stream-en belül. Egy adott felhasználó csak a saját üzenettörténetét láthatja.

Feliratkozhat a streamre, és új üzeneteket kaphat az XREADparancs segítségével. Ez egy kicsit bonyolultabb, mint XRANGE, ezért először egyszerűbb példákkal kezdjük.

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

A fenti példában a nem blokkoló formátum szerepel. XREADFontos megjegyezni, hogy a COUNT opció nem kötelező. Valójában az egyetlen kötelező opció a parancs számára a STREAMS opció, amely a stream-ek listáját jelöli a megfelelő maximális azonosítóval együtt. Írtuk, hogy „STREAMS mystream 0” — az összes mystream azonosítóval rendelkező bejegyzést szeretnénk kapni, amely nagyobb mint „0-0”. Mint a példából látható, a parancs visszaadja a stream nevét, mert több stream-ra is feliratkozhatunk egyszerre. Írhattunk volna például „STREAMS mystream otherstream 0 0”. Vegye figyelembe, hogy a STREAMS opció után először a szükséges stream-ek neveit kell megadnunk, és csak ezután a hozzájuk tartozó azonosítókat.

Ebben az egyszerű formában a parancs nem csinál semmi különöset a XRANGE-hoz képest. Érdekes azonban, hogy könnyen blokkoló parancsá alakíthatjuk, ha megadjuk a BLOCK argumentumot: XREAD A fenti példában egy új BLOCK opció van megadva, 0 milliszekundum várakozási idővel (ez végtelen várakozást jelent). Továbbá, a szokásos azonosító helyett a mystream streamhez egy speciális azonosítót adtak meg. Ez a speciális azonosító azt jelenti, hogy

> XREAD BLOCK 0 STREAMS mystream $
a maximális azonosítót kell használni a mystream streamben. Így csak új üzeneteket fogunk kapni attól a pillanattól kezdve, amikor elkezdtük a figyelést. Bizonyos szempontból ez hasonló az Unix „tail -f” parancsához. XREAD maximális azonosítót kell használnia az mystream folyamatban. Így csak az új üzeneteket kapjuk, a pillanattól kezdve, amikor megkezdtük a hallgatást. Bizonyos értelemben ez hasonló az Unix "tail -f" parancsához.

Kérjük, vegye figyelembe, hogy a BLOCK opció használata esetén nem szükséges külön azonosítót használni $. Bármilyen létező azonosítót felhasználhatunk a folyamban. Ha a parancs azonnal, blokkolás nélkül tudja kiszolgálni a kérésünket, megteszi, különben blokkolva marad.

Blokkoló XREAD több folyamatot is figyelhet egyszerre, csak meg kell adni a nevüket. Ebben az esetben a parancs az első folyamból érkezett rekordot adja vissza. Az első feliratkozó, aki a megadott folyamba le van blokkolva, először fogja megkapni az adatokat.

Fogyasztói Csoportok

Bizonyos feladatok esetén szeretnénk korlátozni a feliratkozók hozzáférését a folyamon belüli üzenetekhez. Egy példa arra, amikor ez hasznos lehet — üzenetsorozat munkásokkal, akik különböző folyami üzeneteket kapnak, lehetővé téve az üzenetek feldolgozásának skálázását.

Ha elképzeljük, hogy három feliratkozónk van: C1, C2, C3, és egy folyamon, amely az 1, 2, 3, 4, 5, 6, 7 üzeneteket tartalmazza, az üzenetek kiszolgálása a következő diagram szerint fog történni:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
E hatás eléréséhez a Redis Stream egy olyan fogalmat használ, amelyet Fogyasztói Csoportnak neveznek. Ez a fogalom hasonló egy pszeudo-feliratkozóhoz, amely adatokat kap a folyamból, de valójában több feliratkozó által van kiszolgálva a csoporton belül, biztosítva bizonyos garanciákat:

Minden egyes üzenet más-más feliratkozóknak kerül kiszállításra a csoporton belül.
A csoporton belül a feliratkozók nevük alapján azonosítják egymást, amely egy regisztrálható karakterlánc. Ha egy feliratkozó ideiglenesen kiesik a csoportból, akkor saját egyedi nevével visszatérhet a csoportba.
Minden Fogyasztói Csoport a "legelső olvasatlan üzenet" elvét követi. Amikor egy feliratkozó új üzeneteket kér, csak azokat az üzeneteket kaphatja meg, amelyeket korábban egyetlen feliratkozónak sem szállítottak le a csoporton belül.
Létezik egy parancs, amely kifejezetten megerősíti, hogy a feliratkozó sikeresen feldolgozta az üzenetet. Amíg ezt a parancsot nem hívják meg, a kért üzenet "pending" (függő) státuszban marad.
A Fogyasztói Csoporton belül minden feliratkozó lekérheti azokat az üzeneteket, amelyeket pontosan neki szállítottak le, de még nem dolgoztak fel ("pending" státuszban).

Bizonyos értelemben a csoport állapotát a következőképpen lehet bemutatni:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" pending üzenetekkel  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" pending üzenetekkel 
|       ... (stb.)                             
+----------------------------------------+

Most itt az ideje, hogy megismerkedjünk a Consumer Group fő parancsaival, nevezetesen:

XGROUP a csoportok létrehozására, megszüntetésére és kezelésére szolgál
XREADGROUP a csoporton keresztül a stream olvasására szolgál
XACK — ez a parancs lehetővé teszi a felhasználónak, hogy megjelölje az üzenetet, mint amely sikeresen feldolgozott

Consumer Group létrehozása

Tegyük fel, hogy a mystream már létezik. A csoport létrehozásának parancsa a következőképpen néz ki:

> XGROUP CREATE mystream mygroup $ OK
A csoport létrehozásakor meg kell adnunk a kezdő azonosítót, ahonnan a csoport üzeneteket kap. Ha csak az összes új üzenetet szeretnénk megkapni, akkor használhatjuk a speciális azonosítót $ (ahogy a fenti példában is látható). Ha a speciális azonosító helyett 0-t adunk meg, akkor a csoport hozzáférhet az összes stream üzenetéhez.

Most, hogy a csoport létrejött, azonnal elkezdhetjük az üzenetek olvasását a parancs segítségével XREADGROUP. Ez a parancs nagyon hasonló a(z) XREAD és támogatja az opcionális BLOCK opciót. Azonban van egy kötelező GROUP opció, amelyet mindig meg kell adni, két érvet: a csoport nevét és a felhasználó nevét. A COUNT opció is támogatott.

Mielőtt olvasnánk a streamet, helyezzünk el néhány üzenetet:

> 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

Most próbáljuk meg olvasni ezt a streamet a csoporton keresztül:

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

A fenti parancs szó szerint a következőket mondja:

«Én, Alice, az mygroup csoport tagja, szeretném olvasni a mystreamből egy olyan üzenetet, amelyet soha nem adtak át senkinek.»

Minden alkalommal, amikor egy feliratkozó műveletet végez egy csoportban, meg kell adnia a nevét, egyértelműen azonosítva magát a csoporton belül. A fenti parancsban van egy nagyon fontos részlet — a speciális azonosító „>”. Ez a speciális azonosító szűri az üzeneteket, csak azokat hagyva meg, amelyeket eddig még nem szállítottak le.

Különleges esetekben valóban megadhat egy valós azonosítót, például 0-t vagy bármilyen más érvényes azonosítót. Ebben az esetben a parancs XREADGROUP visszaadja a „pending” státuszú üzenetek történetét, amelyeket az adott feliratkozónak (Alice) szállítottak, de még nem erősítettek meg a parancs segítségével. XACK.

Ezt a viselkedést azonnal ellenőrizhetjük, ha 0-ás azonosítót adunk meg, az opció nélkül. COUNT. Csak egyetlen várakozó üzenetet fogunk látni, amely az almával kapcsolatos:

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

Ha azonban megerősítjük az üzenetet, mint amelyet sikeresen feldolgoztunk, akkor az már nem fog megjelenni:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (ür lista vagy halmaz)

Most Bob következik, hogy olvasson valamit:

> 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, a mygroup nevű csoport tagja, kért legfeljebb két üzenetet. A parancs csak a nem szállított üzeneteket jelzi a speciális azonosító miatt. Ahogy látható, az „apple” üzenet nem jelenik meg, mivel azt már szállították Alice-nak, így Bob az „orange” és a „strawberry” üzeneteket kapja.

Ebből adódóan Alice, Bob és a csoport bármely más feliratkozója különböző üzeneteket olvashat ugyanabból a folyamból. Emellett a saját fel nem dolgozott üzenetük történetét is olvashatják, vagy az üzeneteket megjelölhetik feldolgozottként.

Van néhány dolog, amire érdemes figyelni:

Amint a feliratkozó elolvassa az üzenetet a parancs segítségével, XREADGROUPez az üzenet „pending” állapotba kerül és rögzül ehhez a konkrét feliratkozóhoz. A csoport többi feliratkozója nem fogja tudni elolvasni ezt az üzenetet.
A feliratkozók automatikusan létrejönnek az első említéskor, nincs szükség a kifejezett létrehozásukra.
A segítségével XREADGROUP Több különböző folytatásból egyidejűleg olvashat üzeneteket, azonban ennek működéséhez először létre kell hoznia csoportokat azonos névvel minden folytatáshoz a következő segítségével XGROUP

Hiba utáni helyreállítás

A felhasználó helyreállíthatja a hibát, és újraolvashatja a „pending” státuszú üzenetlistáját. Azonban a valóságban a felhasználók véglegesen elveszthetik a kapcsolatot. Mi történik a függőben lévő üzenetekkel, ha a felhasználó nem tudja helyreállítani a kapcsolatot a hiba után?
A Fogyasztói Csoport olyan funkciót kínál, amely kifejezetten az ilyen esetekre szolgál – amikor szükséges üzenetek tulajdonosának megváltoztatása.

Első lépésként ki kell adnia a parancsot XPENDING, amely megjeleníti a csoport összes „pending” státuszú üzenetét. A legegyszerűbb formájában a parancs csak két érvet vár: a folytatás nevét és a csoport nevét:

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

A parancs megadta a csoport számára és minden felhasználónak a feldolgozatlan üzenetek számát. Csak Bob van, akinek két feldolgozatlan üzenete van, mert az egyetlen üzenetet, amit Alice kért, már megerősítették a segítségével XACK.

További információt kérhetünk, több érvet használva:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — az azonosítók tartománya (használhatók „-” és „+” jelek)
{count} — a kézbesítési kísérletek száma
{consumer-name} — a csoport neve

> 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

Most már részleteink vannak minden üzenetről: azonosító, felhasználó neve, állásidő milliszekundumokban, és végül a kézbesítési kísérletek száma. Két üzenetünk van Bobtól, és ezek 74170458 milliszekundumot állnak, körülbelül 20 órát.

Vegye figyelembe, hogy senki sem akadályozza meg, hogy ellenőrizzük, mi volt az üzenet tartalma, egyszerűen használva XRANGE.

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

Csak meg kell ismételnie ugyanazt az azonosítót kétszer az érvek között. Most, hogy van némi elképzelésünk, Alice dönthet úgy, hogy Bob után 20 órás állásidő után valószínűleg nem fog helyreállni, és itt az ideje, hogy kérje ezeket az üzeneteket és folytassa a feldolgozásukat Bob helyett. Ehhez a parancsot használjuk XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Ezzel a parancssal megszerezhetjük egy „másik” üzenetet, amelyet még nem dolgoztak fel, a tulajdonos {consumer} megváltoztatásával. Ugyanakkor biztosítani tudjuk a minimális állásidőt {min-idle-time}. Ez segít elkerülni azt a helyzetet, amikor két kliens egyszerre próbálja megváltoztatni a tulajdonost ugyanazon üzenetek esetében:

Ügyfél 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Ügyfél 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Az első ügyfél visszaállítja az állásidőt és növeli a kézbesítési számlálót. Tehát a második ügyfél nem tudja majd igényelni azt.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "üzenet"
      2) "narancs"

Az üzenetet sikeresen igényelte Alice, aki most már feldolgozhatja az üzenetet és megerősítheti azt.

A fenti példa alapján látható, hogy a kérés sikeres végrehajtása visszaadja az üzenet tartalmát. Ez azonban nem feltétlenül szükséges. A JUSTID opció használható az üzenet azonosítók visszaadására. Ez hasznos, ha nem érdeklik az üzenet részletei, és szeretné növelni a rendszer teljesítményét.

Kézbesítési számláló

A számláló, amelyet a kimenetben figyel XPENDING — a minden egyes üzenet kézbesítése. Ez a számláló két módon növekszik: amikor az üzenetet sikeresen igénylik keresztül XCLAIM vagy amikor a hívást használják XREADGROUP.

Teljesen normális, hogy egyes üzenetek többször kerülnek kézbesítésre. A legfontosabb, hogy végül minden üzenet feldolgozásra kerüljön. Néha, amikor az üzenetet feldolgozzák, problémák merülnek fel az üzenet sérülése vagy a feldolgozó kód hibája miatt. Ilyen esetekben előfordulhat, hogy ezt az üzenetet senki sem tudja feldolgozni. Mivel rendelkezünk a kézbesítési kísérletek számlálójával, felhasználhatjuk ezt a számlálót az ilyen helyzetek felfedezésére. Tehát amint a kézbesítési számláló eléri az általad megadott nagy számot, valószínűleg érdemes lenne ezt az üzenetet egy másik folyamba helyezni, és értesítést küldeni a rendszergazdának.

Folyamok állapota

A parancs XINFO különböző információk lekérésére használható a folyamról és csoportjairól. Például a parancs alapvető formája a következőképpen néz ki:

> XINFO STREAM mystream
 1) hossz
 2) (egész szám) 13
 3) radix-fa-kulcsok
 4) (egész szám) 1
 5) radix-fa-csomópontok
 6) (egész szám) 2
 7) csoportok
 8) (egész szám) 2
 9) első-bejegyzés
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) utolsó-bejegyzés
12) 1) 1526569544280-0
    2) 1) "üzenet"
       2) "banán"

A fenti parancs megjeleníti a megadott stream összesített információját. Most egy kicsit bonyolultabb példa:

> XINFO GROUPS mystream
1) 1) név
   2) "mygroup"
   3) fogyasztók
   4) (egész szám) 2
   5) függőben
   6) (egész szám) 2
2) 1) név
   2) "some-other-group"
   3) fogyasztók
   4) (egész szám) 1
   5) függőben
   6) (egész szám) 0

A fenti parancs megjeleníti a megadott stream összes csoportjának információját.

> XINFO CONSUMERS mystream mygroup
1) 1) név
   2) "Alice"
   3) függőben
   4) (egész szám) 1
   5) inaktív
   6) (egész szám) 9104628
2) 1) név
   2) "Bob"
   3) függőben
   4) (egész szám) 1
   5) inaktív
   6) (egész szám) 83841983

A fenti parancs a megadott stream és csoport összes feliratkozójának információját jeleníti meg.
Ha elfelejti a parancs szintaxisát, egyszerűen kérjen segítséget magától a parancstól:

> XINFO HELP
1) XINFO {alparancs} arg arg ... arg. Az alparancsok a következők:
2) CONSUMERS {kulcs} {csoportnév}  -- A {csoportnév} csoport fogyasztóit mutatja.
3) GROUPS {kulcs}                 -- A stream fogyasztói csoportjait mutatja.
4) STREAM {kulcs}                 -- Információt mutat a streamről.
5) HELP                         -- Ezt a segítséget nyomtassa ki.

A stream méretének korlátozása

Sok alkalmazás nem szeretne végtelen ideig adatokat gyűjteni egy streambe. Gyakran hasznos, ha a streamben maximálisan megengedett üzenetek száma is van. Más esetekben érdemes lehet az összes üzenetet áthelyezni a streamből egy másik állandó tárolóba, ha a megadott streamméret elérte a határértéket. A stream méretének korlátozásához használja a MAXLEN paramétert a parancsban: 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
(egész szám) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

A MAXLEN használatakor a régi bejegyzések automatikusan törlődnek, amikor a megadott hosszúságot elérjük, így a stream állandó méretű marad. Azonban a vágás ebben az esetben nem a legtermelékenyebb módon történik a Redis memóriájában. A helyzet javítható az alábbi módon:

XADD mystream MAXLEN ~ 1000 * ... bejegyzésmezők itt ...
A fenti példában a ~ argumentum azt jelenti, hogy nem kell szigorúan korlátoznunk a stream hosszát egy adott értékre. A mi példánkban ez lehet bármilyen szám, amely nagyobb vagy egyenlő 1000-rel (például 1000, 1010 vagy 1030). Csak kifejezetten jeleztük, hogy azt szeretnénk, ha a streamünk legalább 1000 bejegyzést tárolna. Ez sokkal hatékonyabbá teszi a memória használatát a Redisben.

Külön parancs is létezik XTRIM, amely ugyanazt végzi:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Állandó tárolás és replikáció

A Redis Stream asszinkron módon replikálódik a slave csomópontokra, és tárolódik AOF (az összes adat pillanatfelvétele) és RDB (a minden írási művelet naplója) fájlokban. Támogatott a Consumer Groups állapotának replikálása is. Ezért, ha egy üzenet "pending" állapotban van a master csomóponton, akkor a slave csomópontokon ugyanebben az állapotban lesz.

Egyedi elemek eltávolítása a folyamokból

Üzenetek eltávolításához van egy külön parancs XDEL. A parancs megkapja a folyam nevét, amelyet az eltávolítandó üzenetek azonosítói követnek:

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

A parancs használatakor figyelembe kell venni, hogy a memória valójában nem fog azonnal felszabadulni.

Nulla hosszúságú folyamok

A Redis folyamok és más adatstruktúrák közötti különbség az, hogy míg más adatstruktúrák nem tartalmaznak elemeket, ezek a struktúrák eltávolításra kerülnek a memóriából. Például egy rendezett halmaz teljesen eltávolításra kerül, amikor a ZREM utolsó elemet töröl. Ezzel szemben a folyamok lehetnek a memóriában, még akkor is, ha nincsenek elemek benne.