Redis Stream — en ny abstrakt datatyper, der blev introduceret i Redis med version 5.0
Konceptuelt er Redis Stream en liste, hvor du kan tilføje poster. Hver post har en unik identifikator. Som standard genereres identifikatoren automatisk og inkluderer et tidsstempel. Derfor kan du anmode om interval af poster efter tid eller modtage nye data, efterhånden som de kommer ind i strømmen, ligesom Unix-kommandoen "tail -f" læser logfiler og venter på nye data. Bemærk, at flere klienter samtidig kan lytte til strømmen, ligesom mange "tail -f" processer kan læse en fil uden at konflikte med hinanden.

For at forstå alle fordelene ved den nye datatype, lad os hurtigt genopfriske de eksisterende Redis-strukturer, der delvist gentager funktionaliteten af Redis Stream.

Redis PUB/SUB

Redis Pub/Sub er et simpelt beskedsystem, der allerede er indbygget i dit nøgle-værdi-lager. Men simpliciteten har sin pris:

Hvis en udgiver af en eller anden grund går ned, mister den alle sine abonnenter
Udgiveren skal kende den præcise adresse til alle sine abonnenter
Udgiveren kan overbelaste sine abonnenter, hvis dataene offentliggøres hurtigere, end de behandles
Beskeden fjernes fra udgiverens buffer straks efter offentliggørelse, uanset hvor mange abonnenter den er leveret til, og hvor hurtigt de har kunnet behandle beskeden.
Alle abonnenter modtager beskeden samtidig. Abonnenterne skal selv finde ud af, hvordan de skal koordinere rækkefølgen af behandling af den samme besked.
Der er ingen indbygget mekanisme til at bekræfte, at en abonnent har behandlet beskeden succesfuldt. Hvis abonnenten modtager beskeden og går ned under behandlingen, vil udgiveren ikke vide det.

Redis Liste

Redis Liste er en datastruktur, der understøtter læseoperationer med lås. Du kan tilføje og læse beskeder fra starten eller slutningen af listen. På baggrund af denne struktur kan du lave en god stak eller kø til dit distribuerede system, og i de fleste tilfælde vil det være tilstrækkeligt. De grundlæggende forskelle fra Redis Pub/Sub er:

Beskeden leveres til én klient. Den første blokkerede klient, der læser, får dataene først.
Clint skal selv initiere læseoperationen for hvert besked. List ved ikke noget om kunderne.
Beskeder gemmes, indtil de enten bliver læst eller fjernet eksplicit. Hvis du har konfigureret Redis-serveren til at skrive data til disk, øges systemets pålidelighed markant.

Introduktion til Stream

Tilføjelse af en post til stream

Kommandoen XADD tilføjer en ny post til streamen. En post er ikke bare en streng, men består af et eller flere nøgle-værdi-par. Hver post er derfor struktureret og ligner strukturen af en CSV-fil.

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

I eksemplet ovenfor tilføjer vi to felter til streamen med navnet (nøglen) «mystream»: «sensor-id» og «temperature» med værdierne «1234» og «19.8» henholdsvis. Som andet argument accepterer kommandoen en identifikator, der vil blive tildelt posten — denne identifikator identificerer entydigt hver post i streamen. Men i dette tilfælde har vi givet *, fordi vi ønsker, at Redis genererer en ny identifikator for os. Hver ny identifikator vil blive større. Derfor vil hver ny post have en større identifikator i forhold til de tidligere poster.

Formatet for identifikatoren

Identifikatoren for posten, der returneres af kommandoen, XADD, består af to dele:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix-tid i millisekunder (tid server Redis). Hvis den aktuelle tid viser sig at være lig med eller mindre end tiden for den foregående post, bruges tidsstemplingen fra den tidligere post. Derfor, hvis serverens tid går tilbage, vil den nye identifikator stadig have den stigende egenskab.

sequenceNumber bruges til poster, der oprettes inden for samme millisekund. sequenceNumber vil blive øget med 1 i forhold til den foregående post. Da sequenceNumber er 64 bit, bør du i praksis ikke støde på en grænse for antallet af poster, der kan genereres inden for en millisekund.

Formatet for disse identifikatorer kan ved første øjekast virke mærkeligt. En skeptisk læser kunne stille spørgsmålstegn ved, hvorfor tid er en del af identifikatoren. Årsagen er, at Redis-strømme understøtter forespørgsler i intervallet baseret på identifikatorer. Da identifikatoren er knyttet til tidspunktet for oprettelsen af posten, giver det mulighed for at forespørge tidsintervaller. Vi vil se på et konkret eksempel, når vi går videre til studiet af kommandoen XRANGE.

Hvis brugeren af en eller anden grund har brug for at angive sin egen identifikator, der for eksempel er relateret til et eksternt system, kan vi overføre den til kommandoen XADD i stedet for stjernen * som vist nedenfor:

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

Bemærk, at du i dette tilfælde selv skal holde øje med at øge identifikatoren. I vores eksempel er den mindste identifikator "0-1", så kommandoen vil ikke acceptere en anden identifikator, der er lig med eller mindre end "0-1".

> XADD somestream 0-1 foo bar
(error) ERR Den angivne ID i XADD er lig med eller mindre end målkildens øverste element

Antallet af poster i strømmen

Du kan få antallet af poster i strømmen ved blot at bruge kommandoen XLEN. For vores eksempel vil denne kommando returnere følgende værdi:

> XLEN somestream
(integer) 2

Forespørgsler efter interval — XRANGE og XREVRANGE

For at forespørge data efter interval skal vi angive to identifikatorer — begyndelsen og slutningen af intervallet. Det returnerede interval vil inkludere alle elementer, inklusive grænserne. Der er også to specielle identifikatorer "-" og "+", der henholdsvis repræsenterer den mindste (den første post) og den største (den sidste post) identifikator i strømmen. Eksemplet nedenfor vil vise alle poster i strømmen.

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

Hver returneret post repræsenterer et array med to elementer: identifikator og liste over nøgle-værdi-par. Vi har allerede nævnt, at identifikatorerne for poster har relation til tid. Derfor kan vi anmode om et interval for et bestemt tidsrum. Vi kan dog i forespørgslen angive ikke den fulde identifikator, men kun Unix-tid ved at udelade den del, der vedrører sequenceNumber. Den udeladte del af identifikatoren vil automatisk blive sat til nul i starten af området og til den maksimale mulige værdi i slutningen af området. Nedenfor er et eksempel på, hvordan man kan anmode om et interval, der svarer til to millisekunder.

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

Vi har kun én post i dette interval, men i virkelige datasæt kan det returnerede resultat være meget stort. Derfor XRANGE understøtter muligheden COUNT. Ved at angive et antal kan vi blot få de første N poster. Hvis vi skal have de næste N poster (paginering), kan vi bruge den sidst modtagne identifikator, øge den med sequenceNumber en og anmode igen. Lad os se det i det følgende eksempel. Vi begynder at tilføje 10 elementer ved hjælp af XADD (antag, at strømmen mystream allerede var fyldt med 10 elementer). For at begynde iterationen, mens vi modtager 2 elementer pr. anmodning, starter vi med det fulde interval, men med COUNT lige med 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"

For at fortsætte iterationen med de næste to elementer skal vi vælge den sidst modtagne identifikator, nemlig 1519073279157-0, og tilføje 1 til sequenceNumber.
den resulterende identifikator, som i dette tilfælde bliver 1519073279157-1, kan nu bruges som et nyt argument for startområdet til det næste opkald 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"

Og så videre. Da kompleksiteten XRANGE er O(log (N)) for at søge og derefter O(M) for at returnere M elementer, er hvert trin i iterationen hurtigt. På denne måde kan man XRANGE effektivt iterere gennem strømme.

Kommandoen XREVRANGE er ækvivalent med XRANGE, men returnerer elementer i omvendt rækkefølge:

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

Bemærk, at kommandoen XREVRANGE modtager start- og stopområdeargumenter i omvendt rækkefølge.

Læse nye poster ved hjælp af XREAD

Der opstår ofte et behov for at abonnere på en strøm og kun modtage nye meddelelser. Dette koncept kan synes at være lignende Redis Pub/Sub eller blokkerende Redis Liste, men der er grundlæggende forskelle i, hvordan man bruger Redis Stream:

Hver ny meddelelse leveres som standard til hver abonnent. Denne adfærd adskiller sig fra den blokerende Redis-liste, hvor en ny meddelelse kun vil blive læst af én abonnent.
Mens alle meddelelser i Redis Pub/Sub glemmes og aldrig gemmes, gemmes alle meddelelser i Stream i ubestemt tid (medmindre klienten eksplicit anmoder om sletning).
Redis Stream giver mulighed for at opdele adgangen til meddelelser inden for en enkelt stream. En bestemt abonnent kan kun se sin egen personlige historik med meddelelser.

Du kan abonnere på streamen og modtage nye meddelelser ved hjælp af kommandoen XREAD. Dette er lidt mere kompliceret end XRANGE, så vi vil først starte med nogle enklere eksempler.

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

I det ovenstående eksempel er den ikke-blokerende form angivet. XREADBemærk, at COUNT-parameteren ikke er obligatorisk. Faktisk er den eneste obligatoriske parameter for kommandoen STREAMS, som angiver listen over streams sammen med den tilsvarende maksimale identifikator. Vi skrev "STREAMS mystream 0" — vi ønsker at modtage alle poster fra streamen mystream med identifikatoren større end "0-0". Som det fremgår af eksemplet, returnerer kommandoen streamnavnet, fordi vi kan abonnere på flere streams samtidig. Vi kunne for eksempel skrive "STREAMS mystream otherstream 0 0". Bemærk, at vi skal give navnene på alle de ønskede streams først og derefter listen over identifikatorer.

I denne enkle form gør kommandoen ikke noget særligt sammenlignet med XRANGE. Dog er det interessant, at vi nemt kan omdanne XREAD til en blokerende kommando ved at angive argumentet BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
I det ovenstående eksempel er en ny BLOCK-parameter angivet med en tidsgrænse på 0 millisekunder (det betyder uendelig ventetid). Desuden blev der i stedet for at angive en almindelig identifikator for streamen mystream givet en særlig identifikator $. Denne særlige identifikator betyder, at XREAD skal bruge den maksimale identifikator i streamen mystream. Så vi vil kun modtage nye meddelelser fra det øjeblik, vi begyndte at lytte. På en måde ligner det Unix-kommandoen "tail -f".

Bemærk, at vi når vi bruger BLOCK-muligheden ikke nødvendigvis skal anvende et særligt ID $. Vi kan bruge ethvert eksisterende ID i strømmen. Hvis kommandoen kan betjene vores anmodning øjeblikkeligt uden blokering, vil den gøre det; ellers vil den blive blokeret.

Blocker XREAD kan også lytte til flere strømme samtidig, det kræver blot, at deres navne angives. I dette tilfælde vil kommandoen returnere optagelsen fra den første strøm, hvor dataene ankom. Den første abonnent, der er blokeret for den pågældende strøm, vil modtage dataene først.

Forbrugergrupper

I visse opgaver ønsker vi at begrænse adgangen til beskeder fra abonnenter inden for en enkelt strøm. Et eksempel på, hvornår dette kan være nyttigt, er en beskedkø med arbejdere, der modtager forskellige beskeder fra strømmen, hvilket muliggør skaleret behandling af beskeder.

Hvis vi forestiller os, at vi har tre abonnenter C1, C2, C3, og en strøm, der indeholder beskederne 1, 2, 3, 4, 5, 6, 7, vil behandlingen af beskederne ske som vist i diagrammet nedenfor:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
For at opnå denne effekt bruger Redis Stream et koncept kaldet Forbrugergruppe. Dette koncept ligner en pseudo-abonnent, der modtager data fra strømmen, men som faktisk betjenes af flere abonnenter i gruppen, hvilket giver visse garantier:

Hver besked leveres til forskellige abonnenter inden for gruppen.
Inden for gruppen identificeres abonnenterne ved navn, som er en streng med respekt for store og små bogstaver. Hvis en abonnent midlertidigt falder ud af gruppen, kan den genoptages i gruppen med sit eget unikke navn.
Hver forbrugergruppe følger konceptet "første ulæste besked". Når en abonnent anmoder om nye beskeder, kan han kun modtage de beskeder, der aldrig før har været leveret til nogen abonnent inden for gruppen.
Der findes en kommando til eksplicit at bekræfte den succesfulde behandling af beskeden af abonnenten. Indtil denne kommando er kaldt, vil den anmodede besked forblive i status "pending".
Inden for forbrugergruppen kan hver abonnent anmode om historikken for beskeder, der er blevet leveret til ham, men endnu ikke er blevet behandlet (i status "pending").

I en vis forstand kan gruppens tilstand præsenteres som:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" med ventende beskeder  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" med ventende beskeder 
|       ... (og så videre)                             
+----------------------------------------+

Nu er det tid til at blive bekendt med de grundlæggende kommandoer for Consumer Group, nemlig:

XGROUP bruges til at oprette, slette og administrere grupper
XREADGROUP bruges til at læse strømmen gennem gruppen
XACK — denne kommando gør det muligt for abonnenten at markere en besked som behandlet med succes

Oprettelse af Consumer Group

Lad os antage, at strømmen mystream allerede findes. Så vil oprettelseskommandoen se således ud:

> XGROUP Opret mystream mygroup $ OK
Når vi opretter en gruppe, skal vi angive den ID, som gruppen vil begynde at modtage beskeder fra. Hvis vi blot ønsker at modtage alle nye beskeder, kan vi bruge det særlige ID $ (som i vores eksempel ovenfor). Hvis vi i stedet angiver 0 som det særlige ID, vil alle beskeder i strømmen være tilgængelige for gruppen.

Nu hvor gruppen er oprettet, kan vi straks begynde at læse beskeder ved hjælp af kommandoen XREADGROUP. Denne kommando ligner meget XREAD og understøtter den valgfrie mulighed BLOCK. Dog er der en obligatorisk mulighed GROUP, som altid skal angives med to argumenter: gruppens navn og abonnentens navn. Muligheden COUNT understøttes også.

Før vi læser strømmen, lad os tilføje nogle beskeder til den:

> 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

Og nu lad os prøve at læse denne strøm gennem gruppen:

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

Den ovenstående kommando siger bogstaveligt følgende:

"Jeg, abonnenten Alice, medlem af gruppen mygroup, ønsker at læse fra strømmen mystream en besked, som aldrig før er blevet leveret til nogen."

Hver gang en abonnent udfører en handling med en gruppe, skal de angive sit navn, så de entydigt identificerer sig selv inden for gruppen. I den ovenfor nævnte kommando er der en meget vigtig detalje — den særlige identifikator ">". Denne særlige identifikator filtrerer beskeder og efterlader kun dem, der endnu ikke er blevet leveret.

I særlige tilfælde kan du også angive en reel identifikator, som f.eks. 0 eller en hvilken som helst anden gyldig identifikator. I så fald vil kommandoen XREADGROUP vende tilbage med beskeder i status "pending", som er blevet leveret til den angivne abonnent (Alice), men som endnu ikke er blevet bekræftet ved hjælp af kommandoen XACK.

Vi kan kontrollere denne adfærd ved straks at angive identifikatoren 0, uden optionen ANTAL. Vi vil blot se en enkelt ventende besked, nemlig beskeden med æblet:

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

Men hvis vi bekræfter beskeden som vellykket behandlet, vil den ikke længere blive vist:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (tom liste eller sæt)

Nu er det Bobs tur til at læse noget:

> 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, medlem af gruppen mygroup, bad om højst to beskeder. Kommandoen rapporterer kun om ikke-leverede beskeder på grund af den særlige identifikator ">". Som du kan se, vises beskeden "apple" ikke, da den allerede er blevet leveret til Alice, så Bob modtager "orange" og "strawberry."

Derfor kan Alice, Bob og enhver anden abonnent i gruppen læse forskellige beskeder fra den samme strøm. De kan også læse deres historie over ubehandlede beskeder eller markere beskeder som behandlet.

Der er flere ting at have i mente:

Så snart en abonnent læser en besked med kommandoen XREADGROUP, går denne besked ind i en "pending"-tilstand og knyttes til denne specifikke abonnent. Andre abonnenter i gruppen vil ikke kunne læse denne besked.
Abonnenter oprettes automatisk ved første omtale, der er ikke behov for at oprette dem eksplicit.
Ved hjælp af XREADGROUP Du kan læse beskeder fra flere forskellige streams samtidig, men for at det skal fungere, skal du først oprette grupper med samme navn for hver stream ved hjælp af XGROUP

Gendannelse efter nedbrud

Abonnenten kan genoprette sig efter et nedbrud og læse sin liste over beskeder med status "pending". Men i den virkelige verden kan abonnenterne endeligt mislykkes. Hvad sker der med de hængende beskeder for abonnenten, hvis han ikke kan genoprette sig efter et nedbrud?
Consumer Group tilbyder en funktion, der netop bruges i sådanne tilfælde — når det er nødvendigt at skifte ejer af beskeder.

Først og fremmest skal kommandoen kaldes XPENDING, som viser alle beskeder i gruppen med status "pending". I sin simpleste form kaldes kommandoen kun med to argumenter: navnet på streamen og navnet på gruppen:

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

Kommandoen viste antallet af ubehandlede beskeder for hele gruppen og for hver abonnent. Vi har kun Bob med to ubehandlede beskeder, fordi den eneste besked, der blev anmodet af Alice, blev bekræftet med XACK.

Vi kan anmode om flere oplysninger ved at bruge flere argumenter:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — interval af identifikatorer (kan bruge "-" og "+")
{count} — antal leveringsforsøg
{consumer-name} — navn på gruppen

> 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

Nu har vi detaljer for hver besked: identifikator, abonnentens navn, tiden der er gået i millisekunder, og endelig antallet af leveringsforsøg. Vi har to beskeder fra Bob, og de har været hængende i 74170458 millisekunder, cirka 20 timer.

Vær opmærksom på, at der ikke er noget, der forhindrer os i at kontrollere, hvad indholdet af beskeden var, blot ved at bruge XRANGE.

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

Vi skal bare gentage den samme identifikator to gange i argumenterne. Nu, når vi har en idé, kan Alice beslutte, at efter 20 timers ventetid er Bob sandsynligvis ikke i stand til at genoprette sig, og det er tid til at anmode om disse beskeder og genoptage deres behandling i stedet for Bobs. Til dette bruger vi kommandoen XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Ved hjælp af denne kommando kan vi få fat i en "anden" besked, som endnu ikke er blevet behandlet, ved at ændre ejerskabet til {consumer}. Vi kan dog også angive en minimal ventetid {min-idle-time}. Dette hjælper med at undgå situationen, hvor to kunder forsøger samtidigt at ændre ejerskabet af de samme beskeder:

Klient 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Klient 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Den første kunde vil nulstille ventetiden og øge tælleren for antal leveringer. Så den anden kunde vil ikke kunne anmode om den.

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

Besken blev succesfuldt krævet af Alice, som nu kan behandle beskeden og bekræfte den.

Fra det ovenstående eksempel er det tydeligt, at en succesfuld udførelse af anmodningen returnerer indholdet af selve beskeden. Det er dog ikke nødvendigt. JUSTID-muligheden kan bruges til kun at returnere besked-ID'er. Dette er nyttigt, hvis du ikke er interesseret i detaljerne i beskeden og ønsker at forbedre systemets ydelse.

Leveringstæller

Tælleren, som du ser i outputtet XPENDING — er antallet af leveringer af hver besked. Denne tæller stiger på to måder: når beskeden succesfuldt anmodes om via XCLAIM eller når der bruges et kald XREADGROUP.

Det er normalt, at nogle beskeder leveres flere gange. Det vigtigste er, at alle beskeder til sidst bliver behandlet. Nogle gange opstår der problemer, når en besked behandles, enten på grund af beskedens beskadigelse eller fordi behandlingskoden forårsager en fejl. I sådanne tilfælde kan det vise sig, at ingen vil være i stand til at behandle denne besked. Da vi har en tæller for leveringsforsøg, kan vi bruge denne tæller til at opdage sådanne situationer. Derfor, så snart leveringtælleren når et stort antal, som du har angivet, kan det være klogere at placere denne besked i en anden strøm og sende en besked til systemadministratoren.

Stream-tilstande

Kommandoen XINFO bruges til at anmode om forskellige oplysninger om strømmene og deres grupper. For eksempel ser den grundlæggende form af kommandoen således ud:

> XINFO STREAM mystream
 1) længde
 2) (integer) 13
 3) radix-tree-keys
 4) (integer) 1
 5) radix-tree-nodes
 6) (integer) 2
 7) grupper
 8) (integer) 2
 9) første-post
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) sidste-post
12) 1) 1526569544280-0
    2) 1) "message"
       2) "banana"

Den ovenstående kommando viser generel information om den angivne stream. Nu et lidt mere komplekst eksempel:

> XINFO GROUPS mystream
1) 1) navn
   2) "mygroup"
   3) forbrugere
   4) (heltal) 2
   5) ventende
   6) (heltal) 2
2) 1) navn
   2) "some-other-group"
   3) forbrugere
   4) (heltal) 1
   5) ventende
   6) (heltal) 0

Den ovenstående kommando viser generel information om alle grupper i den angivne stream.

> XINFO CONSUMERS mystream mygroup
1) 1) navn
   2) "Alice"
   3) ventende
   4) (heltal) 1
   5) inaktiv
   6) (heltal) 9104628
2) 1) navn
   2) "Bob"
   3) ventende
   4) (heltal) 1
   5) inaktiv
   6) (heltal) 83841983

Den ovenstående kommando viser information om alle abonnenterne i den angivne stream og gruppe.
Hvis du glemmer syntaksen for kommandoen, kan du blot anmode om hjælp ved at bruge selve kommandoen:

> XINFO HELP
1) XINFO {underdel} arg arg ... arg. Underdelene er:
2) CONSUMERS {nøgle} {gruppenavn}  -- Vis forbrugergrupperne af gruppe {gruppenavn}.
3) GROUPS {nøgle}                 -- Vis forbrugergrupperne for streamen.
4) STREAM {nøgle}                 -- Vis information om streamen.
5) HELP                         -- Udskriv denne hjælp.

Begrænsning af streamens størrelse

Mange applikationer ønsker ikke at samle data i en stream for evigt. Det er ofte nyttigt at have det maksimalt tilladte antal meddelelser i en stream. I andre tilfælde er det nyttigt at flytte alle meddelelser fra streamen til et andet permanent lager, når den angivne streamstørrelse er nået. Streamens størrelse kan begrænses ved hjælp af parameteren MAXLEN i kommandoen. XADD:

> XADD mystream MAXLEN 2 * værdi 1
1526654998691-0
> XADD mystream MAXLEN 2 * værdi 2
1526654999635-0
> XADD mystream MAXLEN 2 * værdi 3
1526655000369-0
> XLEN mystream
(heltal) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "værdi"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "værdi"
      2) "3"

Når MAXLEN bruges, fjernes gamle poster automatisk, når den angivne længde er nået, så streamen har en konstant størrelse. Dog sker trimningen ikke den mest effektive måde i hukommelsen i Redis. Situationen kan forbedres på følgende måde:

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...
Argumentet ~ i ovenstående eksempel betyder, at vi ikke nødvendigvis behøver at begrænse streamens længde til en konkret værdi. I vores eksempel kan det være et hvilket som helst tal, der er større end eller lig med 1000 (for eksempel 1000, 1010 eller 1030). Vi har blot eksplicit angivet, at vi ønsker, at vores stream skal gemme mindst 1000 poster. Dette gør hukommelsesarbejdet betydeligt mere effektivt inden for Redis.

Der findes også en separat kommando, XTRIM, der udfører det samme:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Permanent lagring og replikation

Redis Stream replikeres asynkront til slave-noder og gemmes i AOF-filer (snapshot af alle data) og RDB (log over alle skriveoperationer). Der understøttes også replikation af tilstanden for Consumer Groups. Derfor, hvis en besked har status som "pending" på master-noden, vil den have samme status på slave-noderne.

Sletning af enkelte elementer fra strømmen

Der findes en speciel kommando til at slette beskeder XDEL. Kommandoen tager navnet på strømmen, efterfulgt af identifikatorerne for de beskeder, der skal slettes:

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

Når du bruger denne kommando, skal du tage højde for, at hukommelsen faktisk ikke vil blive frigivet straks.

Strømme med nul længde

Forskellen mellem strømme og andre datatyper i Redis er, at når andre datatyper ikke længere har elementer i sig, vil selve datatypen blive slettet fra hukommelsen som en bivirkning. For eksempel vil et sorteret sæt blive helt fjernet, når ZREM fjerner det sidste element. I stedet får strømme lov til at forblive i hukommelsen, selvom der ikke er nogen elementer indeni.