Redis streams, consommateurs et opérations de blocage

Redis a introduit les flux qui imitent la structure des données du journal avec la version 5.0. Stream est une structure de données à consolider uniquement avec un ensemble d'opérations plus riche que dans un fichier journal. Il s'agit de l'un des types de données les plus complexes de Redis car il met en œuvre des opérations de blocage supplémentaires qui permettent aux clients d'attendre les nouvelles données de flux. Ceci est quelque peu similaire au comportement de Redis Pub / Sub ou de listes de blocage, mais des différences fondamentales sont là en ce qui concerne la façon dont les consommateurs consomment les données Redis Stream.

Comme le montre l'illustration précédente, plusieurs avantages par rapport aux listes de redis pub / sub et de blocage peuvent être vues. Chaque nouvel élément de données est livré à chaque consommateur. Contrairement aux listes de suppression de l'élément de liste chaque fois que l'on appelle à BLPOP ou BRPOP, les éléments du flux restent tels qu'ils sont dans le flux. La commande XREAD fonctionne comme un candidat de blocage et non bloquant sur les flux Redis.

La commande xread

La commande xread peut récupérer les entrées à partir de plusieurs flux simultanément tandis que les entrées retournées ont un ID plus grand que le dernier identifiant reçu pour un consommateur donné. Il peut fonctionner à la fois de la manière bloquante et non bloquante. Dans la nature non bloquante, la commande se comporte très similaire à la commande xRange mais avec certaines fonctionnalités supplémentaires répertoriées dans les éléments suivants:

• Il peut récupérer les entrées à partir de l'entrée la plus récente qui a le plus grand identifiant que tout autre élément du flux.
• Il peut lire plusieurs flux en même temps.

Cette commande a une complexité temporelle linéaire lorsque le nombre N d'éléments est stocké dans le flux. Par conséquent, avec un nombre de rendements fixes, la complexité du temps est constante.

La commande xread suit la syntaxe suivante:

Syntaxe:

Xread [count numéro_of_returnd_elements] [Block Blocking_time_in_milliseconds] streams key [key…] id [id…]

COMPTER : Le nombre d'éléments à retourner par la commande. Il limite les lignes retournées à un numéro spécifié.

BLOC : Le temps maximum pour attendre qu'un nouvel élément apparaît dans le flux.

Les deux options précédentes sont facultatives à la commande.

RUISSEAUX : La clé du flux. Il s'agit d'une option obligatoire et doit être la dernière option de la commande car elle accepte la longueur variable des clés et des ID d'entrée.

: L'ID de l'entrée du flux.

Plusieurs touches peuvent être spécifiées car la commande vous permet de lire à partir de plus d'un flux. En même temps, plusieurs ID peuvent être fournies.

Cette commande renvoie une réponse de tableau. Chaque élément de tableau se compose de deux éléments comme indiqué dans le format suivant:

Exemple 1: Inspectez les données météorologiques pour deux emplacements avec XRead non bloquant

Supposons que nous avons obtenu deux flux contenant les données météorologiques pour LA et NYC. Dans notre site de publication de données météorologiques, nous devons consommer à partir des deux flux et récupérer les dernières données météorologiques pour ces deux emplacements. La commande xread est le candidat idéal à utiliser dans ce scénario avec sa variante non bloquante.

Il est temps de créer deux flux nommés Météo: NYC et Météo: LA Et remplissez quelques entrées avec quelques paires de valeurs sur le terrain comme indiqué dans ce qui suit:

Témoire XADD: NYC * vent 45 Humidité 78 Temp 12
Témoire XADD: la * vent 12 Humidité 45 Temp 22

Les deux ruisseaux Météo: NYC et Météo: LC sont créés avec succès et les ID d'entrée retournés sont 1658114094434-0 et 1658114110474-0, respectivement.

Utilisons la commande xread pour lire les deux flux en même temps de manière non bloquante.

Xread Streams Météo: NYC Météo: la 0 0

Comme prévu, la sortie contient les entrées des deux flux avec la séquence d'ID à partir de 0. Il est acceptable de spécifier les ID incomplets comme illustré précédemment où les deux ID sont 0, ce qui est l'horodatage en millisecondes sans la partie numéro de séquence. Par conséquent, la commande précédente peut être écrite comme ce qui suit:

Xread Streams Météo: NYC Météo: LA 0-0 0-0

Ajoutons quelques entrées aux deux flux maintenant.

Témoire XADD: NYC * Vent 10 Humidité 60 Temps 10
xadd météo: la * vent 18 humidité 80 temp 5

Puisque nous avons déjà les derniers ID d'entrée pour les deux flux des commandes précédentes, appelons à nouveau la commande xread pour récupérer toutes les entrées avec des ID plus gros que ceux que nous avons déjà interrogés.

Xread Streams Météo: NYC Météo: LA 1658114094434-0 1658114110474-0

Comme vous pouvez le voir, les ID spécifiés proviennent de la requête précédente. Maintenant, l'appel de commande renvoie toutes les entrées qui ont des ID plus importants que les ID spécifiés.

Comme vous pouvez le voir, les entrées nouvellement ajoutées sont retournées de la commande précédente. Ensuite, ce que vous pouvez faire, c'est prendre les ID d'entrée renvoyés de la commande précédente et appeler le xRead avec ces ID jusqu'à ce que le tableau renvoyé soit vide.

Exemple 2: Obtenez les dernières promotions de pizza avec le blocage de Xread

Il existe une autre variante de la commande xread qui peut être utilisée pour attendre que les éditeurs publient de nouvelles données sur le flux sans se terminer immédiatement comme un appel non bloquant. Supposons un scénario où les gars de la pizza veulent pousser les notifications à tous les clients concernant les dernières promotions disponibles. Il n'y a peut-être pas de promotions à certains jours. Par conséquent, les clients devraient attendre que les nouvelles promos soient disponibles. Il peut être réalisé avec la commande xread avec l'option de bloc en place.

Supposons que la Pizza Company publie les détails promotionnels d'un flux appelé pizzapromos: quotidien. Par conséquent, nous pouvons utiliser la commande xread pour attendre qu'un nouvel élément promotionnel soit ajouté au flux.

Xread Block 50000 Streams PizzapromosNew: Daily $

Dans ce cas, nous spécifions l'ID d'entrée comme $ qui est interprété comme l'ID de saisie supérieure. Par conséquent, la commande interrogera uniquement les nouvelles entrées ajoutées au flux et non aux entrées historiques.

Puisque nous n'avons pas ajouté de nouvelles entrées au flux, il sera dénoncé après 50000 millisecondes avec un néant Retour comme indiqué dans ce qui suit:

Maintenant, ajoutons une entrée au flux en utilisant le XADD tandis qu'un autre consommateur attend les données avec la commande xRead, comme indiqué ci-dessous:

Comme prévu, l'entrée supplémentaire est consommée immédiatement par le consommateur. Depuis l'appel suivant, nous devons nous assurer que nous transmettons l'ID qui est renvoyé de cette commande et non le $. Sinon, nous manquerons les entrées ajoutées entre les deux.

Si plusieurs clients attendent le même flux, les données nouvellement ajoutées sont poussées à tous immédiatement. La commande xread est une commande très utile et recommandée à utiliser pour bloquer les applications de la nature.

Conclusion

Pour résumer, la commande xread est l'une des commandes largement utilisées qui fonctionnent sur les flux redis. Il peut fonctionner de manière à la fois de blocage et de blocage. Comme discuté, la variante non bloquante est très similaire à la commande XRange avec quelques différences. De plus, cette commande peut être utilisée avec l'option Block pour attendre que les éditeurs publient de nouvelles données sur le flux. Dans l'ensemble, la commande xread est spécialisée dans la consommation des données de plusieurs flux simultanément. C'est une fonctionnalité utile que les applications modernes recherchent.