Redis Stream

Redis Stream est une structure de données nouvellement ajoutée dans Redis version 5.0.

Redis Stream est principalement utilisé pour les files d'attente de messages (MQ, Message Queue). Redis lui-même dispose d'une fonctionnalité de publication/abonnement (pub/sub) pour implémenter la fonctionnalité de file d'attente de messages, mais elle présente un inconvénient : les messages ne peuvent pas être persistés. En cas de déconnexion réseau, d'arrêt de Redis, etc., les messages seront perdus.

En termes simples, la publication/abonnement (pub/sub) peut distribuer des messages, mais elle ne peut pas enregistrer l'historique des messages.

Redis Stream assure la persistance des messages et la réplication maître-esclave, permettant à n'importe quel client d'accéder aux données à tout moment, et il peut mémoriser la position d'accès de chaque client, et garantir que les messages ne sont pas perdus.

La structure de Redis Stream est présentée ci-dessous. Elle comporte une liste chaînée de messages qui relie tous les messages ajoutés. Chaque message possède un ID unique et un contenu correspondant :

Chaque Stream possède un nom unique, qui est la clé Redis. Il est automatiquement créé lorsque nous utilisons pour la première fois la commande xadd pour ajouter un message.

Explication du diagramme :

  • Groupe de consommateursGroupe de consommateurs, créé à l'aide de la commande XGROUP CREATE. Un groupe de consommateurs possède plusieurs consommateurs.
  • last_delivered_idCurseur. Chaque groupe de consommateurs possède un curseur, last_delivered_id. Lorsqu'un consommateur lit un message, le curseur last_delivered_id avance.
  • pending_idsVariable d'état du consommateur, utilisée pour maintenir les ID non acquittés du consommateur. pending_ids enregistre les messages qui ont déjà été lus par le client mais qui n'ont pas encore été acked (Acknowledge character : caractère d'acquittement).

Commandes liées aux files d'attente de messages :

  • XADD- Ajouter un message à la fin
  • XTRIM- Tronquer le flux pour limiter sa longueur
  • XDEL- Supprimer un message
  • XLEN- Obtenir le nombre d'éléments contenus dans le flux, c'est-à-dire la longueur du message
  • XRANGE- Obtenir la liste des messages, qui filtre automatiquement les messages supprimés
  • XREVRANGE- Obtenir la liste des messages en ordre inverse, ID du plus grand au plus petit
  • XREAD- Obtenir la liste des messages en mode bloquant ou non bloquant

Commandes liées aux groupes de consommateurs :

  • XGROUP CREATE- Créer un groupe de consommateurs
  • XREADGROUP GROUP- Lire les messages d'un groupe de consommateurs
  • XACK- Marquer un message comme "traité"
  • XGROUP SETID- Définir un nouvel ID de dernier message livré pour le groupe de consommateurs
  • XGROUP DELCONSUMER- Supprimer un consommateur
  • XGROUP DESTROY- Supprimer un groupe de consommateurs
  • XPENDING- Afficher des informations sur les messages en attente
  • XCLAIM- Transférer la propriété d'un message
  • XINFO - Afficher des informations sur les flux et les groupes de consommateurs;
  • XINFO GROUPS- Afficher les informations sur le groupe de consommateurs;
  • XINFO STREAM- Afficher les informations sur le flux

XADD

Utilisez XADD pour ajouter un message à la file d'attente. Si la file d'attente spécifiée n'existe pas, une file d'attente sera créée. Format de syntaxe XADD :

XADD key ID field value [field value ...]
  • keyNom de la file d'attente; créez-la si elle n'existe pas
  • IDID du message. Nous utilisons * pour indiquer qu'il est généré par Redis. Il peut être personnalisé, mais vous devez garantir son caractère incrémental.
  • field valueEnregistrement.

Exemple

redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1601372323627-1"
redis> XLEN mystream
(entier) 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

Utilisez XTRIM pour tronquer le flux afin de limiter sa longueur. Format de syntaxe :

XTRIM key MAXLEN [~] count
  • keyNom de la file d'attente
  • MAXLENLongueur
  • countNombre

Exemple

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
(entier) 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

Utilisez XDEL pour supprimer des messages. Format de syntaxe :

XDEL key ID [ID ...]
  • keyNom de la file d'attente
  • IDID du message

Exemple

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(entier) 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

Utilisez XLEN pour obtenir le nombre d'éléments contenus dans le flux, c'est-à-dire la longueur des messages. Format de syntaxe :

XLEN key
  • keyNom de la file d'attente

Exemple

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

XRANGE

Utilisez XRANGE pour obtenir la liste des messages, qui filtre automatiquement les messages supprimés. Format de syntaxe :

XRANGE key start end [COUNT count]
  • keyNom de la file d'attente
  • startValeur de début,-représente la valeur minimale
  • endValeur de fin,+représente la valeur maximale
  • countNombre

Exemple

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
(entier) 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

Utilisez XREVRANGE pour obtenir la liste des messages, qui filtre automatiquement les messages supprimés. Format de syntaxe :

XREVRANGE key end start [COUNT count]
  • keyNom de la file d'attente
  • endValeur de fin,+représente la valeur maximale
  • startValeur de début,-représente la valeur minimale
  • countNombre

Exemple

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
(entier) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"
   2) 1) "name"
      2) "Ngozi"
      3) "surname"
      4) "Adichie"
redis>

XREAD

Utilisez XREAD pour obtenir la liste des messages en mode bloquant ou non bloquant. Format de syntaxe :

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
  • countNombre
  • millisecondsFacultatif. Millisecondes de blocage. Si non défini, le mode est non bloquant.
  • keyNom de la file d'attente
  • idID du message

Exemple

# Lire deux messages depuis la tête du 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

Utilisez XGROUP CREATE pour créer un groupe de consommateurs. Format de syntaxe :

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
  • keyNom de la file d'attente; créez-la si elle n'existe pas
  • groupnameNom du groupe.
  • $Indique une consommation depuis la queue, en n'acceptant que les nouveaux messages. Tous les messages actuels du Stream seront ignorés.

Commencer à consommer depuis la tête :

XGROUP CREATE mystream consumer-group-name 0-0

Commencer à consommer depuis la queue :

XGROUP CREATE mystream consumer-group-name $

XREADGROUP GROUP

Utilisez XREADGROUP GROUP pour lire des messages depuis un groupe de consommateurs. Format de syntaxe :

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • groupNom du groupe de consommateurs
  • consumerNom du consommateur.
  • countNombre à lire.
  • millisecondsMillisecondes de blocage.
  • keyNom de la file d'attente.
  • IDID du message.
XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
Autres extensions