Redis Stream

Redis Stream es una estructura de datos añadida en la versión 5.0 de Redis.

Redis Stream se utiliza principalmente para colas de mensajes (MQ, Message Queue). Redis en sí tiene una función de publicación/suscripción (pub/sub) para implementar colas de mensajes, pero tiene la desventaja de que los mensajes no se pueden persistir. Si ocurre una desconexión de red, una caída de Redis, etc., los mensajes se perderán.

En pocas palabras, la publicación/suscripción (pub/sub) puede distribuir mensajes, pero no puede registrar mensajes históricos.

Redis Stream ofrece persistencia de mensajes y replicación maestro-respaldo, lo que permite a cualquier cliente acceder a los datos en cualquier momento, recordar la posición de acceso de cada cliente y garantizar que no se pierdan mensajes.

La estructura de Redis Stream es la siguiente: tiene una lista enlazada de mensajes que conecta todos los mensajes añadidos. Cada mensaje tiene un ID único y su contenido correspondiente:

Cada Stream tiene un nombre único, que es la clave de Redis, y se crea automáticamente cuando usamos el comando xadd para añadir mensajes por primera vez.

Análisis de la imagen anterior:

  • Consumer GroupGrupo de consumidores, creado mediante el comando XGROUP CREATE. Un grupo de consumidores tiene varios consumidores.
  • last_delivered_idCursor. Cada grupo de consumidores tiene un cursor last_delivered_id. Cuando cualquier consumidor lee un mensaje, el cursor last_delivered_id avanza.
  • pending_idsVariable de estado del consumidor, cuya función es mantener los IDs no confirmados del consumidor. pending_ids registra los mensajes que ya han sido leídos por el cliente pero que aún no han sido confirmados con ack (carácter de confirmación).

Comandos relacionados con la cola de mensajes:

  • XADDAñadir mensaje al final
  • XTRIMRecortar el stream para limitar su longitud
  • XDELEliminar mensaje
  • XLENObtener el número de elementos del stream, es decir, la longitud de los mensajes
  • XRANGEObtener la lista de mensajes, filtrando automáticamente los mensajes eliminados
  • XREVRANGEObtener la lista de mensajes en orden inverso, con los ID de mayor a menor
  • XREADObtener la lista de mensajes en modo bloqueante o no bloqueante

Comandos relacionados con grupos de consumidores:

  • XGROUP CREATECrear un grupo de consumidores
  • XREADGROUP GROUPLeer mensajes de un grupo de consumidores
  • XACKMarcar un mensaje como "procesado"
  • XGROUP SETIDEstablecer un nuevo ID de último mensaje entregado para el grupo de consumidores
  • XGROUP DELCONSUMEREliminar un consumidor
  • XGROUP DESTROYEliminar un grupo de consumidores
  • XPENDINGMostrar información sobre mensajes pendientes
  • XCLAIMTransferir la propiedad de los mensajes
  • XINFO Ver información sobre el stream y los grupos de consumidores
  • XINFO GROUPSMostrar información del grupo de consumidores
  • XINFO STREAMMostrar información del stream

XADD

Utilice XADD para añadir mensajes a la cola. Si la cola especificada no existe, se creará. Formato de sintaxis de XADD:

XADD key ID field value [field value ...]
  • keyNombre de la cola; si no existe, se crea
  • IDID del mensaje. Usamos * para indicar que Redis lo genera automáticamente. También puede personalizarse, pero debe garantizarse su incrementalidad.
  • field valueRegistro

Ejemplo

redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1601372323627-1"
redis> XLEN mystream
(integer) 2
redis> XRANGE mystream - +
1) 1) "1601372323627-0"
   2) 1) "name"
      2) "Sara"
      3) "surname"
      4) "OConnor"
2) 1) "1601372323627-1"
   2) 1) "field1"
      2) "value1"
      3) "field2"
      4) "value2"
      5) "field3"
      6) "value3"
redis>

XTRIM

Utilice XTRIM para recortar el stream y limitar su longitud. Formato de sintaxis:

XTRIM key MAXLEN [~] count
  • keyNombre de la cola
  • MAXLENLongitud
  • countCantidad

Ejemplo

127.0.0.1:6379> XADD mystream * field1 A field2 B field3 C field4 D
"1601372434568-0"
127.0.0.1:6379> XTRIM mystream MAXLEN 2
(integer) 0
127.0.0.1:6379> XRANGE mystream - +
1) 1) "1601372434568-0"
   2) 1) "field1"
      2) "A"
      3) "field2"
      4) "B"
      5) "field3"
      6) "C"
      7) "field4"
      8) "D"
127.0.0.1:6379>

redis>

XDEL

Utilice XDEL para eliminar mensajes. Formato de sintaxis:

XDEL key ID [ID ...]
  • keyNombre de la cola
  • IDID del mensaje

Ejemplo

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(integer) 1
127.0.0.1:6379> XRANGE mystream - +
1) 1) 1538561698944-0
   2) 1) "a"
      2) "1"
2) 1) 1538561701744-0
   2) 1) "c"
      2) "3"

XLEN

Utilice XLEN para obtener el número de elementos del stream, es decir, la longitud de los mensajes. Formato de sintaxis:

XLEN key
  • keyNombre de la cola

Ejemplo

redis> XADD mystream * item 1
"1601372563177-0"
redis> XADD mystream * item 2
"1601372563178-0"
redis> XADD mystream * item 3
"1601372563178-1"
redis> XLEN mystream
(integer) 3
redis>

XRANGE

Utilice XRANGE para obtener la lista de mensajes, que filtrará automáticamente los mensajes eliminados. Formato de sintaxis:

XRANGE key start end [COUNT count]
  • keyNombre de la cola
  • startValor inicial,-indica el valor mínimo
  • endValor final,+indica el valor máximo
  • countCantidad

Ejemplo

redis> XADD writers * name Virginia surname Woolf
"1601372577811-0"
redis> XADD writers * name Jane surname Austen
"1601372577811-1"
redis> XADD writers * name Toni surname Morrison
"1601372577811-2"
redis> XADD writers * name Agatha surname Christie
"1601372577812-0"
redis> XADD writers * name Ngozi surname Adichie
"1601372577812-1"
redis> XLEN writers
(integer) 5
redis> XRANGE writers - + COUNT 2
1) 1) "1601372577811-0"
   2) 1) "name"
      2) "Virginia"
      3) "surname"
      4) "Woolf"
2) 1) "1601372577811-1"
   2) 1) "name"
      2) "Jane"
      3) "surname"
      4) "Austen"
redis>

XREVRANGE

Utilice XREVRANGE para obtener la lista de mensajes, que filtrará automáticamente los mensajes eliminados. Formato de sintaxis:

XREVRANGE key end start [COUNT count]
  • keyNombre de la cola
  • endValor final,+indica el valor máximo
  • startValor inicial,-indica el valor mínimo
  • countCantidad

Ejemplo

redis> XADD writers * name Virginia surname Woolf
"1601372731458-0"
redis> XADD writers * name Jane surname Austen
"1601372731459-0"
redis> XADD writers * name Toni surname Morrison
"1601372731459-1"
redis> XADD writers * name Agatha surname Christie
"1601372731459-2"
redis> XADD writers * name Ngozi surname Adichie
"1601372731459-3"
redis> XLEN writers
(integer) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"
   2) 1) "name"
      2) "Ngozi"
      3) "surname"
      4) "Adichie"
redis>

XREAD

Utilice XREAD para obtener la lista de mensajes en modo bloqueante o no bloqueante. Formato de sintaxis:

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
  • countCantidad
  • millisecondsOpcional, milisegundos de bloqueo. Si no se establece, es modo no bloqueante
  • keyNombre de la cola
  • idID del mensaje

Ejemplo

# Leer dos mensajes desde el principio del Stream
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"
   2) 1) 1) 1526984818136-0
         2) 1) "duration"
            2) "1532"
            3) "event-id"
            4) "5"
            5) "user-id"
            6) "7782813"
      2) 1) 1526999352406-0
         2) 1) "duration"
            2) "812"
            3) "event-id"
            4) "9"
            5) "user-id"
            6) "388234"
2) 1) "writers"
   2) 1) 1) 1526985676425-0
         2) 1) "name"
            2) "Virginia"
            3) "surname"
            4) "Woolf"
      2) 1) 1526985685298-0
         2) 1) "name"
            2) "Jane"
            3) "surname"
            4) "Austen"

XGROUP CREATE

Utilice XGROUP CREATE para crear un grupo de consumidores. Formato de sintaxis:

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
  • keyNombre de la cola; si no existe, se crea
  • groupnameNombre del grupo
  • $Indica que se comienza a consumir desde el final, aceptando solo mensajes nuevos. Todos los mensajes actuales del Stream se ignorarán.

Consumir desde el principio:

XGROUP CREATE mystream consumer-group-name 0-0

Consumir desde el final:

XGROUP CREATE mystream consumer-group-name $

XREADGROUP GROUP

Utilice XREADGROUP GROUP para leer mensajes del grupo de consumidores. Formato de sintaxis:

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • groupNombre del grupo de consumidores
  • consumerNombre del consumidor
  • countCantidad a leer
  • millisecondsMilisegundos de bloqueo
  • keyNombre de la cola
  • IDID del mensaje
XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
Otras extensiones