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 registrazioni. Ogni registrazione ha un identificatore unico. Di default, l'identificatore viene generato automaticamente e include un timestamp. Puoi quindi richiedere intervalli di registrazioni nel tempo o ricevere nuovi dati man mano che arrivano nel flusso, proprio come il comando Unix «tail -f» legge un file di log e attende nuovi dati. Nota che più client possono ascoltare il flusso contemporaneamente, proprio come molti processi «tail -f» possono leggere un file senza conflitti.

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

Redis PUB/SUB

Redis Pub/Sub è un semplice sistema di messaggistica già integrato nel tuo sistema di archiviazione key-value. Tuttavia, per la sua semplicità si paga un prezzo:

Se l'editore per qualche motivo smette di funzionare, perde tutti i suoi iscritti
L'editore deve conoscere l'indirizzo esatto di tutti i suoi iscritti
L'editore può sovraccaricare i propri abbonati se i dati vengono pubblicati più rapidamente di quanto possano essere 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 hanno potuto elaborarlo.
Tutti gli abbonati riceveranno il messaggio contemporaneamente. Gli abbonati devono concordare tra loro l'ordine di elaborazione dello stesso messaggio.
Non esiste un meccanismo integrato per confermare l'avvenuta elaborazione del messaggio da parte dell'abbonato. Se l'abbonato riceve il messaggio e si blocca durante l'elaborazione, l'editore non ne sarà a conoscenza.

Redis List

Redis List è una struttura dati che supporta comandi di lettura con blocco. Puoi aggiungere e leggere messaggi dall'inizio o dalla fine della lista. Su questa struttura si può realizzare un buon stack o una coda per il tuo sistema distribuito, e 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 che si blocca in lettura riceverà per primo i dati.
Clint deve avviare lui stesso l'operazione di lettura di ogni messaggio. List non conosce i clienti.
I messaggi vengono conservati fino a quando non vengono letti o esplicitamente eliminati. Se hai configurato il server Redis per scrivere i dati su disco, l'affidabilità del sistema aumenta notevolmente.

Introduzione a 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. Pertanto, 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, aggiungiamo al flusso chiamato «mystream» due campi: «sensor-id» e «temperature» con i valori «1234» e «19.8» rispettivamente. Come secondo argomento, il comando accetta un identificatore, che sarà assegnato alla registrazione: questo identificatore identifica univocamente ogni registrazione nel flusso. Tuttavia, in questo caso, abbiamo passato *, poiché vogliamo che Redis generi un nuovo identificatore per noi. Ogni nuovo identificatore verrà incrementato. Pertanto, ogni nuova registrazione avrà un identificatore maggiore rispetto alle registrazioni precedenti.

Formato identificatore

L'identificatore della registrazione, restituito dal comando XADD, è composto da due parti:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — tempo Unix in millisecondi (tempo server Redis). Tuttavia, se il tempo attuale è uguale o minore rispetto al tempo della registrazione precedente, verrà utilizzato il timestamp della registrazione precedente. Pertanto, se il tempo del server torna indietro, il nuovo identificatore manterrà comunque la sua caratteristica di incremento.

sequenceNumber viene utilizzato per le registrazioni create nello stesso millisecondo. sequenceNumber verrà incrementato di 1 rispetto alla registrazione precedente. Poiché sequenceNumber ha una dimensione di 64 bit, quindi nella pratica non dovresti incontrare limitazioni 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 richieste di intervallo sugli identificatori. Poiché l’identificatore è legato al momento della creazione del record, questo consente di richiedere intervalli di tempo. Esamineremo un esempio specifico quando inizieremo a studiare il comando XRANGE.

Se per qualche motivo l'utente ha bisogno di specificare un proprio identificatore, che ad esempio è associato a un sistema esterno, possiamo passarlo al comando XADD invece del simbolo * 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 devi tenere traccia dell'incremento dell'identificatore. Nel nostro esempio, l’identificatore minimo è "0-1", quindi il comando non accetterà un altro identificatore che sia pari o minore di "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 — l'inizio e la fine dell'intervallo. L'intervallo restituito includerà tutti gli elementi, comprese le estremità. Ci sono 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 sotto mostrerà 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 è un array di due elementi: un identificatore e un elenco di coppie chiave-valore. Abbiamo già discusso del fatto che gli identificatori dei record sono legati al tempo. Pertanto, possiamo richiedere un intervallo di tempo specifico. Tuttavia, possiamo specificare nella richiesta non l'intero identificatore, ma solo il tempo Unix, omettendo la parte relativa a sequenceNumber. La parte omessa dell'identificatore verrà automaticamente impostata a zero all'inizio dell'intervallo e al valore massimo alla fine dell'intervallo. Di seguito è riportato un esempio su 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, incrementando il suo sequenceNumber per unit e richiedere di nuovo. Diamo un'occhiata a questo nel prossimo esempio. Iniziamo ad aggiungere 10 elementi utilizzando XADD (supponiamo che il flusso mystream sia già stato riempito con 10 elementi). Per iniziare l'iterazione, ricevendo 2 elementi alla volta, partiamo dall'intero intervallo, ma con COUNT uguale 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, quindi 1519073279157-0, e sommare 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 passo di iterazione è veloce. Pertanto, con XRANGE è possibile iterare efficacemente i flussi.

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

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

Si prega di notare che il comando XREVRANGE accetta argomenti di intervallo start e stop in ordine inverso.

Leggere nuovi record usando XREAD

Spesso sorge la necessità di iscriversi a un flusso e ricevere solo nuovi messaggi. Questo concetto potrebbe 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, per impostazione predefinita, viene consegnato a ciascun abbonato. Questo comportamento è diverso da una lista Redis bloccante, dove un nuovo messaggio sarà letto solo da un singolo abbonato.
Mentre in Redis Pub/Sub tutti i messaggi vengono dimenticati e mai salvati, nello Stream tutti i messaggi vengono conservati indefinitamente (a meno che il cliente non richieda esplicitamente la loro cancellazione).
Redis Stream consente di delimitare l'accesso ai messaggi all'interno di un singolo flusso. Un abbonato specifico può vedere solo la propria cronologia di messaggi.

Puoi iscriverti a un flusso e ricevere nuovi messaggi usando il comando XREAD. È un po' più complicato di XRANGE, quindi inizieremo prima con esempi più semplici.

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

Nell'esempio sopra è indicata la forma non bloccante XREAD. Si noti che l'opzione COUNT non è obbligatoria. Infatti, l'unica opzione obbligatoria del comando è l'opzione STREAMS, che specifica l'elenco dei flussi insieme all'identificatore massimo corrispondente. Abbiamo scritto «STREAMS mystream 0» — vogliamo ricevere tutti i record del flusso mystream con identificatore maggiore di «0-0». Come si può vedere dall'esempio, il comando restituisce il nome del flusso, perché possiamo iscriverci a più flussi contemporaneamente. Potremmo scrivere, ad esempio, «STREAMS mystream otherstream 0 0». Si noti che dopo l'opzione STREAMS dobbiamo prima fornire i nomi di tutti i flussi desiderati e solo successivamente 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 specificata la nuova opzione BLOCK con un tempo di attesa di 0 millisecondi (questo significa attesa infinita). Inoltre, anziché passare un normale identificatore per il flusso mystream, è stato passato un identificatore speciale $. Questo identificatore speciale indica che XREAD deve utilizzare come identificatore il massimo identificatore nel flusso mystream. Quindi 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 noti che quando si utilizza l'opzione BLOCK non è necessario utilizzare l'identificatore speciale $. Possiamo utilizzare qualsiasi identificatore esistente nel flusso. Se il comando sarà in grado di gestire la nostra richiesta immediatamente, senza bloccarsi, lo farà; altrimenti, si bloccherà.

Bloccante XREAD può anche ascoltare più flussi contemporaneamente, basta specificare i loro nomi. In questo caso, il comando restituirà il record del primo flusso in cui sono arrivati i dati. Il primo subscriber bloccato per tale flusso riceverà i dati per primo.

Gruppi di Consumo

In alcune situazioni, vogliamo limitare l'accesso dei sottoscrittori ai messaggi all'interno di un singolo flusso. Un esempio di quando questo può essere utile è una coda di messaggi con lavoratori che riceveranno messaggi diversi dal flusso, consentendo di scalare l'elaborazione dei messaggi.

Se immaginiamo di avere tre sottoscrittori C1, C2, C3 e un flusso che contiene i messaggi 1, 2, 3, 4, 5, 6, 7, il servizio 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 Consumo. Questo concetto è simile a un pseudo-sottoscrittore che riceve dati dal flusso, ma è effettivamente servito da più sottoscrittori all'interno del gruppo, fornendo certe garanzie:

Ogni messaggio viene consegnato a diversi sottoscrittori all'interno del gruppo.
All'interno del gruppo, i sottoscrittori sono identificati dal loro nome, che è una stringa sensibile al caso. Se un sottoscrittore esce temporaneamente dal gruppo, può riprendersi in esso utilizzando il proprio nome univoco.
Ogni Consumer Group segue il concetto di "primo messaggio non letto". Quando un sottoscrittore richiede nuovi messaggi, può ricevere solo quei messaggi che non sono stati mai precedentemente consegnati a nessun sottoscrittore all'interno del gruppo.
Esiste un comando che conferma esplicitamente il trattamento riuscito del messaggio da parte del sottoscrittore. Finché non viene chiamato questo comando, il messaggio richiesto rimarrà in stato di "pending".
All'interno della Consumer Group, ogni sottoscrittore può richiedere la cronologia dei messaggi che sono stati consegnati a lui, ma che non sono stati ancora elaborati (in stato di "pending").

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

+----------------------------------------+
| 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 è il momento di conoscere i comandi fondamentali per la Consumer Group, cioè:

XGROUP utilizzato per creare, distruggere e gestire i gruppi
XREADGROUP utilizzato per leggere il flusso tramite il gruppo
XACK — è un 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 avrà questo aspetto:

> XGROUP CREATE mystream mygroup $ OK
Quando creiamo un gruppo dobbiamo fornire l'ID da cui il gruppo inizierà a ricevere i messaggi. Se vogliamo semplicemente ricevere tutti i nuovi messaggi, possiamo utilizzare un ID speciale $ (come nel nostro esempio precedente). Se invece forniamo 0 al posto dell'ID speciale, il gruppo avrà accesso a tutti i messaggi del flusso.

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

Prima di iniziare a leggere il flusso, mettiamo alcuni messaggi al suo interno:

> 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

Ora proviamo a leggere questo stream attraverso un 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 dice letteralmente quanto segue:

«Io, Alice, iscritta, membro del gruppo mygroup, voglio leggere dallo stream mystream un messaggio che non è mai stato consegnato a nessuno precedentemente».

Ogni volta che un'iscritto esegue un'operazione con il gruppo, deve specificare il proprio nome, identificandosi in modo univoco all'interno del gruppo. Nel comando sopra riportato c'è un altro dettaglio molto importante — un identificatore speciale «>». Questo identificatore speciale filtra i messaggi, lasciando solo quelli che non sono stati mai consegnati.

Inoltre, in casi particolari, puoi specificare un identificatore reale, come 0 o qualsiasi altro identificatore valido. In questo caso, il comando XREADGROUP ti restituirà la storia dei messaggi con stato "pending" che sono stati recapitati al destinatario specificato (Alice), ma non sono stati ancora confermati tramite il comando XACK.

Possiamo verificare questo comportamento specificando subito l'ID 0, senza opzione CONTEGGIO. Vedremo solo un singolo 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) "message"
            2) "apple"

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

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)

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) "message"
            2) "orange"
      2) 1) 1526569506935-0
         2) 1) "message"
            2) "strawberry"

Bob, membro del gruppo mygroup, ha richiesto non più di due messaggi. Il comando restituisce solo i messaggi non recapitati a causa dell'ID speciale ">". Come puoi vedere, il messaggio "apple" non appare, poiché è già stato recapitato ad Alice, quindi Bob riceve "orange" e "strawberry".

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

Ci sono alcune cose da tenere a mente:

Non appena un abbonato considera un messaggio come un comando XREADGROUP, questo messaggio passa in uno stato di "in attesa" e viene assegnato a quel specifico abbonato. Altri abbonati del gruppo non potranno leggere questo messaggio.
Gli abbonati vengono creati automaticamente alla prima menzione, non è necessario crearli esplicitamente.
Con XREADGROUP puoi leggere messaggi da più flussi diversi contemporaneamente, tuttavia, affinché questo funzioni, è necessario creare in anticipo gruppi con lo stesso nome per ogni flusso utilizzando XGROUP

Ripristino dopo un guasto

Un abbonato può riprendersi dopo un guasto e rileggere la propria lista di messaggi con stato "in attesa". Tuttavia, nel mondo reale, gli abbonati possono fallire definitivamente. Cosa succede ai messaggi in attesa di un abbonato se non riesce a recuperarsi dopo un guasto?
Consumer Group offre una funzione utilizzabile proprio in questi casi: quando è necessario cambiare il proprietario dei messaggi.

Per prima cosa, è necessario invocare il comando XPENDING, che mostra tutti i messaggi del gruppo con stato "pending". Nella sua forma più semplice, il comando viene richiamato solo con due argomenti: il nome del flusso e il nome del gruppo:

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

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

Possiamo richiedere ulteriori informazioni, utilizzando più argomenti:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — intervallo di identificatori (è possibile utilizzare "-" e "+")
{count} — numero di tentativi di consegna
{consumer-name} — nome del gruppo

> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-0
   2) "Bob"
   3) (intero) 74170458
   4) (intero) 1
2) 1) 1526569506935-0
   2) "Bob"
   3) (intero) 74170458
   4) (intero) 1

Ora abbiamo dettagli per ogni messaggio: identificatore, nome dell'abbonato, tempo di inattività in millisecondi e, infine, il numero di tentativi di consegna. Abbiamo due messaggi da Bob, che sono fermi da 74170458 millisecondi, circa 20 ore.

Si noti 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 negli argomenti. Ora che abbiamo un'idea di base, Alice può decidere che dopo 20 ore di inattività Bob probabilmente non si riprenderà, ed è tempo di richiedere questi messaggi e riprendere il loro trattamento al posto di Bob. Per fare ciò, utilizziamo il comando XCLAIM:

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

Cliente 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Cliente 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Il primo cliente ridurrà il tempo di inattività e aumenterà il conteggio delle consegne. Così, il secondo cliente non potrà richiederlo.

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

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. Questo è utile se non si sono interessati ai dettagli del messaggio e si desidera aumentare le prestazioni del sistema.

Contatore delle consegne

Il contatore che osservi nell'output XPENDING è il numero di consegne per ciascun messaggio. Questo contatore aumenta in due modi: quando il messaggio è stato 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 elaborati. A volte, in fase di elaborazione di un messaggio, si verificano problemi a causa di un danneggiamento del messaggio stesso o un errore nel codice del gestore di messaggi. In tal caso, potrebbe accadere che nessuno sia in grado di elaborare questo messaggio. Poiché abbiamo un contatore dei tentativi di consegna, possiamo utilizzare questo contatore per rilevare tali situazioni. Pertanto, non appena il contatore delle consegne raggiunge un numero elevato da te impostato, è probabile che sia più saggio spostare tale messaggio in un altro flusso e inviare una notifica all'amministratore di sistema.

Stato dei flussi

Team XINFO è utilizzato per richiedere diverse informazioni sui flussi e sui loro gruppi. Ad esempio, la forma base del comando appare come segue:

> 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-voce
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) ultima-voce
12) 1) 1526569544280-0
    2) 1) "messaggio"
       2) "banana"

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

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

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

> XINFO CONSUMERS mystream mygroup
1) 1) nome
   2) "Alice"
   3) in sospeso
   4) (intero) 1
   5) inattivo
   6) (intero) 9104628
2) 1) nome
   2) "Bob"
   3) in sospeso
   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, chiedi semplicemente aiuto al comando stesso:

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

Limite di dimensione del flusso

Molti applicativi non desiderano raccogliere dati in un flusso per sempre. Spesso è utile avere il numero massimo consentito di messaggi nel flusso. In altri casi, è utile spostare tutti i messaggi dal flusso in un'altra memoria permanente una volta raggiunta una dimensione specificata per il flusso. Puoi 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
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

Quando utilizzi MAXLEN, le voci più vecchie vengono automaticamente eliminate al raggiungimento della lunghezza specificata, quindi il flusso ha una dimensione fissa. Tuttavia, il taglio in questo caso non avviene nel modo più efficace in memoria Redis. La situazione può essere migliorata nel seguente modo:

XADD mystream MAXLEN ~ 1000 * ... campi di inserimento 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 vogliamo che il nostro flusso conservi almeno 1000 record. Questo rende la gestione della memoria molto più efficiente all'interno di Redis.

Esiste anche un comando separato XTRIM, che esegue la stessa operazione:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Archiviazione persistente e replica

Redis Stream viene replicato in modo asincrono sui nodi slave e salvato in file di tipo AOF (istantanea di tutti i dati) e RDB (log di tutte le operazioni di scrittura). La replicazione dello stato dei Consumer Groups è anch'essa supportata. Pertanto, se un messaggio è nello stato "in attesa" sul nodo master, su quelli slave avrà lo stesso stato.

Cancellazione di singoli elementi dal flusso

Per rimuovere i messaggi esiste un comando specifico XDEL. Il comando prende il nome del flusso seguito dagli identificatori dei messaggi da eliminare:

> 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, è importante notare che la memoria non verrà liberata immediatamente.

Stream di lunghezza zero

La differenza tra gli stream 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 il comando ZREM elimina l'ultimo elemento. Al contrario, gli stream possono rimanere in memoria anche senza elementi al loro interno.