Redis Stream — un nuovo tipo di dato astratto, introdotto in Redis con il rilascio della versione 5.0
Concettualmente, Redis Stream è una lista in cui puoi aggiungere record. Ogni record ha un identificatore unico. Per impostazione predefinita, l'identificatore viene generato automaticamente e include un timestamp. Pertanto, puoi richiedere intervalli di record in base al tempo o ricevere nuovi dati man mano che arrivano nel flusso, proprio come il comando Unix «tail -f» legge un file di log e si arresta in attesa di nuovi dati. Si noti che più client possono ascoltare contemporaneamente il flusso, proprio come molti processi «tail -f» possono leggere un file senza conflittare tra loro.

Per comprendere tutti i vantaggi del nuovo tipo di dato, ricordiamo brevemente le strutture Redis esistenti da tempo che ripetono parzialmente la funzionalità di Redis Stream.

Redis PUB/SUB

Redis Pub/Sub è un semplice sistema di messaggistica, già integrato nel tuo archivio key-value. Tuttavia, per la sua semplicità bisogna pagare il prezzo:

Se l'editore si guasta per qualsiasi motivo, perde tutti i suoi abbonati
L'editore deve conoscere l'indirizzo esatto di tutti i suoi abbonati
L'editore può sovraccaricare il lavoro dei suoi abbonati se i dati vengono pubblicati più velocemente di quanto vengano elaborati
Il messaggio viene rimosso dal buffer dell'editore subito dopo la pubblicazione, indipendentemente dal numero di abbonati a cui è stato consegnato e dalla velocità con cui questi hanno potuto elaborare il messaggio.
Tutti gli abbonati riceveranno il messaggio contemporaneamente. Gli abbonati devono concordare tra loro l'ordine di elaborazione dello stesso messaggio.
Non c'è un meccanismo integrato di conferma dell'elaborazione riuscita del messaggio da parte dell'abbonato. Se l'abbonato riceve il messaggio e si guasta durante l'elaborazione, l'editore non ne verrà a conoscenza.

Redis List

Redis List è una struttura dati che supporta comandi di lettura bloccati. Puoi aggiungere e leggere messaggi dall'inizio o dalla fine della lista. Su questa base, puoi creare un buon stack o una coda per il tuo sistema distribuito e questo nella maggior parte dei casi sarà sufficiente. Le principali differenze rispetto a Redis Pub/Sub:

Il messaggio viene consegnato a un solo cliente. Il primo cliente bloccato nella lettura riceverà prima i dati.
Clint deve avviare l'operazione di lettura di ogni messaggio da solo. List non sa nulla sui clienti.
I messaggi vengono memorizzati fino a quando qualcuno non li legge o non li elimina esplicitamente. Se hai configurato il server Redis per scrivere i dati su disco, l'affidabilità del sistema aumenta drasticamente.

Introduzione allo Stream

Aggiunta di un record allo stream

Team XADD aggiunge un nuovo record allo stream. Un record non è semplicemente una stringa, ma è composto da una o più coppie chiave-valore. In questo modo, ogni record è già strutturato e ricorda la struttura di un file CSV.

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

Nell'esempio sopra, stiamo aggiungendo allo stream con il nome (chiave) "mystream" due campi: "sensor-id" e "temperature" con i valori "1234" e "19.8" rispettivamente. Come secondo argomento, il comando accetta un identificatore che verrà assegnato al record: questo identificatore identifica in modo univoco ciascun record nello stream. Tuttavia, in questo caso abbiamo passato *, perché vogliamo che Redis generi un nuovo identificatore per noi. Ogni nuovo identificatore aumenterà. Pertanto, ogni nuovo record avrà un identificatore maggiore rispetto ai record precedenti.

Formato dell'identificatore

L'identificatore del record, restituito dal comando XADD, è composto da due parti:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — tempo Unix in millisecondi (tempo server Redis). Tuttavia, se l'ora corrente risulta essere uguale o minore rispetto al tempo del record precedente, verrà utilizzato il timestamp del record precedente. Quindi, se il tempo del server torna indietro, il nuovo identificatore manterrà comunque la proprietà di aumentare.

sequenceNumber viene utilizzato per i record creati nello stesso millisecondo. sequenceNumber verrà aumentato di 1 rispetto al record precedente. Poiché sequenceNumber ha una dimensione di 64 bit, nella pratica non dovresti incontrare un limite sul numero di record che possono essere generati in un millisecondo.

Il formato di tali identificatori può sembrare strano a prima vista. Un lettore scettico potrebbe chiedersi perché il tempo faccia parte dell'identificatore. La ragione è che i flussi Redis supportano le richieste di intervallo per gli identificatori. Poiché l'identificatore è legato al momento di creazione del record, si ha la possibilità di richiedere intervalli temporali. Esamineremo un esempio specifico quando passeremo allo studio del comando XRANGE.

Se per qualche motivo l'utente ha bisogno di specificare un proprio identificatore, che ad esempio è collegato a un sistema esterno, possiamo passarne uno al comando XADD anziché il segno * come mostrato di seguito:

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

Si noti che in questo caso è necessario seguire personalmente l'incremento dell'identificatore. Nel nostro esempio, l'identificatore minimo è "0-1", quindi il comando non accetterà un altro identificatore che sia uguale o inferiore a "0-1".

> XADD somestream 0-1 foo bar
(error) ERR L'ID specificato in XADD è uguale o minore dell'elemento superiore del flusso di destinazione

Numero di record nel flusso

È possibile ottenere il numero di record nel flusso semplicemente utilizzando il comando XLEN. Per il nostro esempio, questo comando restituirà il seguente valore:

> XLEN somestream
(integer) 2

Richieste per intervallo — XRANGE e XREVRANGE

Per richiedere dati per intervallo, dobbiamo specificare due identificatori: inizio e fine dell'intervallo. L'intervallo restituito includerà tutti gli elementi, inclusi i confini. Esistono anche due identificatori speciali «-» e «+», che significano rispettivamente il più piccolo (il primo record) e il più grande (l'ultimo record) identificatore nel flusso. L'esempio qui sotto restituirà tutti i record del flusso.

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

Ogni record restituito rappresenta un array di due elementi: identificatore e lista di coppie chiave-valore. Abbiamo già detto che gli identificatori dei record hanno a che fare con il tempo. Pertanto, possiamo richiedere un intervallo di un intervallo temporale specifico. Tuttavia, possiamo specificare nella richiesta non un identificatore completo, ma solo il tempo Unix, saltando la parte relativa a sequenceNumber. La parte mancante dell'identificatore sarà automaticamente impostata a zero all'inizio dell'intervallo e al valore massimo possibile alla fine dell'intervallo. Qui di seguito è riportato un esempio di come richiedere un intervallo di due millisecondi.

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

Abbiamo solo un record in questo intervallo, tuttavia, nei set di dati reali, il risultato restituito può essere enorme. Per questo motivo XRANGE supporta l'opzione COUNT. Specificando un numero, possiamo semplicemente ottenere i primi N record. Se abbiamo bisogno di ottenere i successivi N record (paginazione), possiamo utilizzare l'ultimo identificatore ricevuto, incrementarlo sequenceNumber di uno e richiedere nuovamente. Diamo un'occhiata a questo nel seguente esempio. Iniziamo ad aggiungere 10 elementi utilizzando XADD (presupponiamo che il flusso mystream sia già stato riempito con 10 elementi). Per iniziare a iterare, recuperando 2 elementi per comando, iniziamo con l'intervallo completo, ma con COUNT impostato a 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"

Per continuare l'iterazione con i successivi due elementi, dobbiamo selezionare l'ultimo identificatore ricevuto, cioè 1519073279157-0, e aggiungere 1 a sequenceNumber.
L'identificatore risultante, in questo caso 1519073279157-1, può ora essere utilizzato come nuovo argomento di inizio intervallo per la prossima chiamata 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"

E così via. Poiché la complessità XRANGE è O(log (N)) per la ricerca, e poi O(M) per restituire M elementi, ogni passaggio dell'iterazione è rapido. Così, con XRANGE è possibile iterare i flussi in modo efficiente.

Team XREVRANGE è equivalente XRANGE, ma restituisce gli elementi in ordine inverso:

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

Nota che il comando XREVRANGE accetta gli argomenti di intervallo start e stop in ordine inverso.

Leggere nuovi record utilizzando XREAD

Spesso si presenta la necessità di iscriversi a un flusso e ricevere solo nuovi messaggi. Questo concetto può sembrare simile a Redis Pub/Sub o a una lista Redis bloccante, ma ci sono differenze fondamentali su come utilizzare Redis Stream:

Ogni nuovo messaggio viene consegnato per impostazione predefinita a ogni abbonato. Questo comportamento è diverso da una lista Redis bloccante, dove un nuovo messaggio verrà letto solo da un singolo abbonato.
Mentre in Redis Pub/Sub tutti i messaggi vengono dimenticati e non vengono mai salvati, in Stream tutti i messaggi vengono conservati per un periodo indefinito (a meno che il cliente non richieda esplicitamente la cancellazione).
Redis Stream consente di segregare l'accesso ai messaggi all'interno di un singolo stream. Un abbonato specifico può vedere solo la propria storia personale di messaggi.

Puoi iscriverti a uno stream e ricevere nuovi messaggi utilizzando il comando XREAD. È un po' più complicato di XRANGE, quindi iniziamo con esempi più semplici.

> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1519073278252-0
         2) 1) "foo"
            2) "value_1"
      2) 1) 1519073279157-0
         2) 1) "foo"
            2) "value_2"

Nell'esempio sopra è indicata una forma non bloccante. XREADSi noti che l'opzione COUNT non è obbligatoria. In effetti, l'unica opzione obbligatoria per il comando è l'opzione STREAMS, che definisce l'elenco degli stream insieme al rispettivo identificatore massimo. Abbiamo scritto "STREAMS mystream 0" — vogliamo ricevere tutte le voci dello stream mystream con un identificatore maggiore di "0-0". Come si vede nell'esempio, il comando restituisce il nome dello stream, perché possiamo iscrivici a più stream contemporaneamente. Potremmo scrivere, ad esempio, "STREAMS mystream otherstream 0 0". Si noti che dopo l'opzione STREAMS dobbiamo prima fornire i nomi di tutti gli stream necessari e solo allora l'elenco degli identificatori.

In questa forma semplice, il comando non fa nulla di particolare rispetto a XRANGE. Tuttavia, è interessante notare che possiamo facilmente trasformare XREAD in un comando bloccante specificando l'argomento BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
Nell'esempio sopra, è stata indicata una nuova opzione BLOCK con un timeout di 0 millisecondi (ciò significa attesa infinita). Inoltre, invece di passare l'identificatore normale per lo stream mystream, è stato passato un identificatore speciale $. Questo identificatore speciale indica che XREAD deve utilizzare come identificatore l'identificatore massimo dello stream mystream. Così riceveremo solo nuovi messaggi a partire dal momento in cui abbiamo iniziato ad ascoltare. In un certo senso, è simile al comando Unix "tail -f".

Si prega di notare che, quando si utilizza l'opzione BLOCK, non è necessario utilizzare un identificatore speciale $. Possiamo usare qualsiasi identificatore esistente nel flusso. Se il comando può elaborare la nostra richiesta immediatamente, senza blocchi, lo farà; altrimenti, sarà in attesa.

Bloccante XREAD può anche ascoltare più flussi contemporaneamente, è sufficiente specificarne i nomi. In questo caso, il comando restituirà i dati del primo flusso in cui sono arrivati.

Gruppi di Consumer

In alcune situazioni vogliamo delimitare l'accesso degli iscritti ai messaggi all'interno di un flusso. Un esempio in cui questo può essere utile è una coda di messaggi con lavoratori che riceveranno messaggi diversi dal flusso, permettendo quindi di scalare l'elaborazione dei messaggi.

Se immaginiamo di avere tre iscritti C1, C2, C3 e un flusso che contiene i messaggi 1, 2, 3, 4, 5, 6, 7, la gestione dei messaggi avverrà come mostrato nel diagramma qui sotto:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Per ottenere questo effetto, Redis Stream utilizza un concetto chiamato Gruppo di Consumer. Questo concetto è simile a un pseudo-iscritto, che riceve dati dal flusso ma è effettivamente gestito da più iscritti all'interno del gruppo, fornendo determinate garanzie:

Ogni messaggio è consegnato a diversi iscritti all'interno del gruppo.
All'interno del gruppo, gli iscritti sono identificati da un nome che è una stringa sensibile al maiuscolo. Se un iscritto esce temporaneamente dal gruppo, può riprendersi nel gruppo con il proprio nome unico.
Ogni Gruppo di Consumer segue il concetto di "primo messaggio non letto". Quando un iscritto richiede nuovi messaggi, può ricevere solo quei messaggi che non sono mai stati precedentemente consegnati a nessun iscritto all'interno del gruppo.
Esiste un comando per confermare esplicitamente che un messaggio è stato elaborato con successo dall'iscritto. Fino a quando questo comando non viene chiamato, il messaggio richiesto rimarrà in stato "in attesa".
All'interno del Gruppo di Consumer, ogni iscritto può richiedere lo storico dei messaggi che gli sono stati consegnati, ma che non sono ancora stati elaborati (in stato "in attesa").

In un certo senso, lo stato del gruppo può essere rappresentato come:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" con messaggi in attesa  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" con messaggi in attesa 
|       ... (e così via)                             
+----------------------------------------+

Ora è tempo di conoscere i comandi fondamentali per il Consumer Group, ovvero:

XGROUP viene utilizzato per creare, distruggere e gestire i gruppi
XREADGROUP viene utilizzato per leggere il flusso tramite il gruppo
XACK è il comando che consente all'abbonato di contrassegnare un messaggio come elaborato con successo

Creazione del Consumer Group

Supponiamo che il flusso mystream esista già. Allora il comando per creare il gruppo sarà:

> CREA XGROUP mystream mygroup $ OK
Quando creiamo un gruppo dobbiamo passare l'identificatore da cui il gruppo inizierà a ricevere i messaggi. Se vogliamo semplicemente ricevere tutti i nuovi messaggi, possiamo usare un identificatore speciale $ (come nel nostro esempio precedente). Se invece indichiamo 0 al posto dell'identificatore speciale, tutti i messaggi del flusso saranno disponibili per il gruppo.

Ora che il gruppo è stato creato, possiamo iniziare immediatamente a leggere i messaggi utilizzando il comando XREADGROUP. Questo comando è molto simile a XREAD e supporta l'opzione facoltativa BLOCK. Tuttavia, c'è un'opzione obbligatoria GROUP, che deve sempre essere specificata con due argomenti: il nome del gruppo e il nome dell'abbonato. Anche l'opzione COUNT è supportata.

Prima di leggere il flusso, mettiamo lì alcuni messaggi:

> 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

E ora proviamo a leggere questo flusso tramite il gruppo:

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

Il comando sopra riportato afferma di seguito:

«Io, Alice-abbonato, membro del gruppo mygroup, voglio leggere dal flusso mystream un messaggio che non è mai stato consegnato a nessuno prima».

Ogni volta che un abbonato esegue un'operazione con il gruppo, deve indicare il proprio nome, identificandosi in modo univoco all'interno del gruppo. Un'ulteriore importante dettaglio nel comando sopra riportato è l'identificatore speciale «>». Questo identificatore speciale filtra i messaggi, lasciando solo quelli che finora non sono stati consegnati.

Inoltre, in casi speciali, puoi specificare un identificatore reale, come 0 o qualsiasi altro identificatore valido. In questo caso il comando XREADGROUP restituirà la cronologia dei messaggi con lo stato «pending», che sono stati consegnati all'abbonato indicato (Alice), ma non sono ancora stati confermati tramite il comando XACK.

Possiamo verificare questo comportamento specificando subito l'identificatore 0, senza opzione VARIANZA. Vedremo solo un messaggio in attesa, cioè il messaggio con la mela:

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "messaggio"
            2) "mela"

Tuttavia, se confermiamo il messaggio come elaborato con successo, non verrà più visualizzato:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (lista o insieme vuoto)

Ora è il turno di Bob di leggere qualcosa:

> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569498055-0
         2) 1) "messaggio"
            2) "arancia"
      2) 1) 1526569506935-0
         2) 1) "messaggio"
            2) "fragola"

Bob, membro del gruppo mygroup, ha richiesto non più di due messaggi. Il comando fornisce solo i messaggi non consegnati a causa dell'identificatore speciale «>». Come puoi vedere, il messaggio «mela» non viene visualizzato, poiché è già stato consegnato ad Alice, quindi Bob riceve «arancia» e «fragola».

In questo modo, Alice, Bob e qualsiasi altro abbonato al gruppo possono leggere messaggi diversi dallo stesso stream. Possono anche leggere la propria cronologia di messaggi non elaborati o contrassegnare i messaggi come elaborati.

Ci sono alcune cose da tenere a mente:

Una volta che un abbonato considera un messaggio con il comando XREADGROUP, questo messaggio passa allo stato «pending» e viene assegnato a questo specifico abbonato. Altri abbonati al gruppo non potranno leggere questo messaggio.
Gli abbonati vengono creati automaticamente al primo utilizzo, non è necessario crearli esplicitamente.
Utilizzando XREADGROUP puoi leggere messaggi da più flussi diversi contemporaneamente, tuttavia, affinché ciò funzioni, devi prima creare gruppi con lo stesso nome per ogni flusso utilizzando XGROUP

Ripristino dopo un guasto

Il consumatore può riprendersi da un guasto e rileggere la sua lista di messaggi con stato "pending". Tuttavia, nel mondo reale, i consumatori possono fallire definitivamente. Cosa succede ai messaggi in attesa del consumatore se non riesce a riprendersi dal guasto?
Il Consumer Group offre una funzione pensata proprio per tali casi - quando è necessario cambiare il proprietario dei messaggi.

Innanzitutto, è necessario chiamare il comando XPENDING, che visualizza tutti i messaggi del gruppo con stato "pending". Nella sua forma più semplice, il comando viene chiamato con solo due argomenti: il nome del flusso e il nome del gruppo:

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

Il comando ha restituito il numero di messaggi non elaborati per l'intero gruppo e per ciascun consumatore. Abbiamo solo Bob con due messaggi non elaborati, poiché l'unico messaggio richiesto da Alice è stato confermato con XACK.

Possiamo richiedere informazioni aggiuntive utilizzando più argomenti:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — intervallo di identificatori (puoi usare «-» e «+»)
{count} — numero di tentativi di consegna
{consumer-name} — nome del gruppo

> 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

Ora abbiamo i dettagli per ogni messaggio: identificatore, nome del consumatore, tempo di inattività in millisecondi e, infine, numero di tentativi di consegna. Abbiamo due messaggi da Bob, che sono in attesa da 74170458 millisecondi, circa 20 ore.

Nota che nulla ci impedisce di controllare quale fosse il contenuto del messaggio, semplicemente utilizzando XRANGE.

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

Dobbiamo semplicemente ripetere lo stesso identificatore due volte tra gli argomenti. Ora, avendo un’idea, Alice può decidere che dopo 20 ore di inattività Bob probabilmente non si riprenderà, ed è giunto il momento di richiedere questi messaggi e riprendere il loro trattamento anziché quello di Bob. A tal fine utilizziamo il comando XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Con questo comando possiamo ottenere un messaggio "esterno" che non è ancora stato elaborato, cambiando il proprietario in {consumer}. Tuttavia, possiamo anche fornire un tempo di inattività minimo {min-idle-time}. Questo aiuta a evitare situazioni in cui due clienti tentano contemporaneamente di cambiare il proprietario dello stesso messaggio:

Cliente 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Cliente 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Il primo cliente azzererà il tempo di inattività e aumenterà il contatore delle consegne. Quindi il secondo cliente non potrà richiederlo.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "messaggio"
      2) "arancio"

Il messaggio è stato richiesto con successo da Alice, che ora può elaborarlo e confermarlo.

Dall'esempio sopra, si può notare che l'esecuzione riuscita della richiesta restituisce il contenuto stesso del messaggio. Tuttavia, questo non è obbligatorio. L'opzione JUSTID può essere utilizzata per restituire solo gli identificatori del messaggio. Ciò è utile se non si è interessati ai dettagli del messaggio e si desidera aumentare le prestazioni del sistema.

Contatore di consegna

Il contatore che osservi nell'output XPENDING — è il numero di consegne di ciascun messaggio. Questo contatore aumenta in due modi: quando il messaggio viene richiesto con successo tramite XCLAIM o quando viene effettuata una chiamata XREADGROUP.

È normale che alcuni messaggi vengano consegnati più volte. L'importante è che alla fine tutti i messaggi siano stati elaborati. A volte, durante l'elaborazione di un messaggio, si verificano problemi a causa della corruzione del messaggio stesso o l'elaborazione del messaggio provoca un errore nel codice del gestore. In tal caso, potrebbe succedere che nessuno sia in grado di elaborare quel messaggio. Poiché abbiamo un contatore di tentativi di consegna, possiamo utilizzare questo contatore per rilevare tali situazioni. Pertanto, una volta che il contatore delle consegne raggiunge un numero elevato stabilito da te, probabilmente sarà più saggio spostare quel messaggio in un altro flusso e inviare una notifica all'amministratore di sistema.

Stato dei flussi

Team XINFO è utilizzato per richiedere varie informazioni sul flusso e sui suoi gruppi. Ad esempio, la forma di base del comando è la seguente:

> XINFO STREAM mystream
 1) lunghezza
 2) (intero) 13
 3) chiavi-albero-radice
 4) (intero) 1
 5) nodi-albero-radice
 6) (intero) 2
 7) gruppi
 8) (intero) 2
 9) prima-vocazione
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) ultima-vocazione
12) 1) 1526569544280-0
    2) 1) "messaggio"
       2) "banana"

Il comando sopra visualizza informazioni generali sul flusso specificato. Ora un esempio leggermente più complesso:

> XINFO GROUPS mystream
1) 1) nome
   2) "mygroup"
   3) consumatori
   4) (intero) 2
   5) in attesa
   6) (intero) 2
2) 1) nome
   2) "some-other-group"
   3) consumatori
   4) (intero) 1
   5) in attesa
   6) (intero) 0

Il comando sopra visualizza informazioni su tutti i gruppi del flusso specificato

> XINFO CONSUMERS mystream mygroup
1) 1) nome
   2) "Alice"
   3) in attesa
   4) (intero) 1
   5) inattivo
   6) (intero) 9104628
2) 1) nome
   2) "Bob"
   3) in attesa
   4) (intero) 1
   5) inattivo
   6) (intero) 83841983

Il comando sopra visualizza informazioni su tutti gli abbonati del flusso e del gruppo specificati.
Se dimentichi la sintassi del comando, puoi semplicemente chiedere aiuto al comando stesso:

> XINFO HELP
1) XINFO {sottocomando} arg arg ... arg. I sottocomandi sono:
2) CONSUMERS {chiave} {nomegruppo}  -- Mostra i gruppi di consumo del gruppo {nomegruppo}.
3) GROUPS {chiave}                 -- Mostra i gruppi di consumo del flusso.
4) STREAM {chiave}                 -- Mostra informazioni sul flusso.
5) HELP                         -- Stampa questa guida.

Limitazione della dimensione del flusso

Molte applicazioni non vogliono raccogliere dati in un flusso per sempre. È spesso utile avere un numero massimo di messaggi nel flusso. In altri casi, è utile spostare tutti i messaggi dal flusso in un altro archivio permanente una volta raggiunta una certa dimensione del flusso. È possibile limitare la dimensione del flusso utilizzando il parametro MAXLEN nel comando 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
(intero) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

Utilizzando MAXLEN, le vecchie registrazioni vengono automaticamente eliminate al raggiungimento della lunghezza specificata, quindi il flusso ha una dimensione costante. Tuttavia, l'accorciamento in questo caso non avviene nel modo più efficiente nella memoria di Redis. Si può migliorare la situazione nel seguente modo:

XADD mystream MAXLEN ~ 1000 * ... campi di voce qui ...
L'argomento ~ nell'esempio sopra significa che non è necessario limitare la lunghezza del flusso a un valore specifico. Nel nostro esempio può essere qualsiasi numero maggiore o uguale a 1000 (ad esempio, 1000, 1010 o 1030). Abbiamo semplicemente specificato che desideriamo che il nostro flusso conservi almeno 1000 registrazioni. Questo rende l'uso della memoria molto più efficiente all'interno di Redis.

C'è anche un comando separato XTRIM, che esegue la stessa operazione:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Archiviazione permanente e replica

Redis Stream si replica in modo asincrono sui nodi slave e viene salvato in file di tipo AOF (istantanea di tutti i dati) e RDB (log di tutte le operazioni di scrittura). Supporta anche la replicazione dello stato dei Consumer Groups. Quindi, se un messaggio è in stato "pending" sul nodo master, anche sui nodi slave quel messaggio avrà lo stesso stato.

Rimozione di singoli elementi dal flusso

Per rimuovere messaggi esiste un comando speciale XDEL. Il comando riceve il nome del flusso, seguito dagli identificatori dei messaggi che devono essere rimossi:

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

Quando si utilizza questo comando, è necessario tenere presente che la memoria non sarà liberata immediatamente.

Flussi di lunghezza zero

La differenza tra flussi e altre strutture dati di Redis è che, quando altre strutture dati non contengono più elementi, come effetto collaterale, la struttura stessa viene rimossa dalla memoria. Ad esempio, un insieme ordinato verrà completamente rimosso quando una chiamata a ZREM elimina l'ultimo elemento. Invece, ai flussi è consentito rimanere in memoria anche senza elementi al loro interno.