Les flux de données en temps réel sont au cœur de nombreuses applications modernes, des systèmes de messagerie instantanée aux tableaux de bord de surveillance. Redis Stream offre une structure de données efficace et persistante pour gérer ces flux de manière ordonnée et évolutive.

Modèle de stockage interne

Chaque message dans un flux est identifié de manière unique par un identfiiant composé d'un horodatage en millisecondes et d'un numéro de séquence, par exemple 1634523456789-0. Le contenu du message est constitué de paires champ-valeur. En interne, Redis organise ces messages dans une liste chaînée qui préserve l'ordre chronologique. Cette structure est persistée via les mécanismes AOF et RDB de Redis.

La fonctionnalité des groupes de consommateurs permet à plusieurs consommateurs de lire un même flux de manière coordonnée. Lorsqu'un message est traité par un consommateur au sein d'un groupe, il est marqué comme consommé pour ce groupe. Chaque groupe maintient un curseur interne suivant la progression de la lecture. Ce modèle est particulièrement adapté aux architectures distribuées.

Commandes fondamentales

Ajout de messages

La commande XADD insère un nouveau message. L'opérateur * génère automatiquement un ID unique. L'option MAXLEN peut limiter la taille du flux, avec une suppression graduelle des anciens messages (~ pour une suppression approximative).

XADD flux_evenements * type "connexion" utilisateur "alice" date "2024-01-01"

Lecture de messages

Pour lire une plage de messages, on utilise XRANGE avec les bornes - (début) et + (fin). XRANGE est utile pour récupérer un historique.

XRANGE flux_evenements - + COUNT 20

La commande XREAD permet de lire les messages entrants. En mode bloquant (BLOCK), elle attend de nouveaux messages pendant une durée spécifiée.

XREAD BLOCK 5000 STREAMS flux_evenements $

Gestion des groupes

Pour créer un groupe de consommateurs :

XGROUP CREATE flux_evenements groupe_analyse 0

Un consommateur peut lire les messages non traités dans son groupe avec XREADGROUP.

XREADGROUP GROUP groupe_analyse worker1 COUNT 10 STREAMS flux_evenements >

Pour inspecter les messages en attente d'acquittement :

XPENDING flux_evenements groupe_analyse - + 10

Applications typiques

Système de messagerie : Chaque conversation peut être stockée comme un flux distinct. Les clients de l'utilisateur (mobile, web) s'inscrivent en tant que consommateurs d'un même groupe pour synchroniser l'état des messages lus.

Supervision d'infrastructure : Des métriques (CPU, mémoire) sont publiées dans des flux par agent. Les différents services d'analyse consomment ces données en parallèle via des groupes pour générer des alertes ou des rapports statistiques.

Architectures événementielles : Les événements métier (commande passée, paiement reçu) alimentent des flux. Les différents services back-end s'abonnent et traitent les événements qui les concernent, garantissant ainsi un découplage et une scalabilité horizontale.

Implémentation avec Go

Voici un exemple d'utilisation du client go-redis pour manipuler des flux.

Initialisation de la connexion

package main

import (
	"context"
	"fmt"
	"github.com/redis/go-redis/v9"
)

var backgroundCtx = context.Background()

func createClient() *redis.Client {
	return redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "",
		DB:       0,
	})
}

func main() {
	client := createClient()
	defer client.Close()

	status := client.Ping(backgroundCtx)
	if status.Err() != nil {
		panic(status.Err())
	}
	fmt.Println("Connecté à Redis")
}

Publication et récupération d'événements

func ajouterEvenement(client *redis.Client, cleFlux string, donnees map[string]interface{}) (string, error) {
	args := &redis.XAddArgs{
		Stream: cleFlux,
		Values: donnees,
	}
	return client.XAdd(backgroundCtx, args).Result()
}

func lireHistorique(client *redis.Client, cleFlux string, limite int64) ([]redis.XMessage, error) {
	args := &redis.XRangeArgs{
		Stream: cleFlux,
		Min:    "-",
		Max:    "+",
		Count:  limite,
	}
	return client.XRange(backgroundCtx, args).Result()
}

// Exemple d'utilisation
idMsg, err := ajouterEvenement(client, "journal_actions", map[string]interface{}{
	"action":   "upload",
	"fichier":  "rapport.pdf",
	"utilisateur": "bob",
})

messages, err := lireHistorique(client, "journal_actions", 10)

Consommation avec un groupe

func initialiserGroupe(client *redis.Client, cleFlux, nomGroupe string) error {
	return client.XGroupCreateMkStream(backgroundCtx, cleFlux, nomGroupe, "0").Err()
}

func consommerAvecGroupe(client *redis.Client, cleFlux, nomGroupe, nomConsommateur string, nombre int64) ([]redis.XStream, error) {
	args := &redis.XReadGroupArgs{
		Group:    nomGroupe,
		Consumer: nomConsommateur,
		Streams:  []string{cleFlux, ">"},
		Count:    nombre,
	}
	return client.XReadGroup(backgroundCtx, args).Result()
}

// Exemple d'utilisation
err := initialiserGroupe(client, "taches_a_faire", "equipe_dev")
fluxResultats, err := consommerAvecGroupe(client, "taches_a_faire", "equipe_dev", "dev1", 5)