Redis Pub/Sub vs Streams : lequel choisir en Python ?

Redis Pub/Sub et Redis Streams semblent répondre au même besoin : un producteur émet, des consommateurs reçoivent. Les deux s’installent en quelques lignes, et sur un environnement de développement, ils donnent le même résultat.

La différence apparaît avec une seule question : que se passe-t-il si le consommateur n’est pas là au moment où le message part ? Un déploiement, une coupure réseau ou un worker qui plante suffisent à la poser. La réponse sépare en réalité trois modèles : la diffusion Pub/Sub, la lecture d’un journal avec XREAD, et le partage du travail avec les groupes de consommateurs.

En bref :

• Pub/Sub : pour des signaux éphémères, un message manqué est perdu.
• XREAD : pour une diffusion rejouable, chaque lecteur garde son curseur.
• Groupes de consommateurs : pour répartir le travail entre workers, avec reprise après crash.

Redis Pub/Sub : une diffusion sans mémoire

Avec Pub/Sub, le producteur publie sur un canal et Redis transmet immédiatement le message aux clients abonnés à cet instant :

from redis.asyncio import Redis

redis = Redis.from_url("redis://localhost:6379", decode_responses=True)


async def publish_status(order_id: int, payload: str) -> int:
    return await redis.publish(f"orders:{order_id}", payload)


async def listen(order_id: int) -> None:
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"orders:{order_id}")
    async for message in pubsub.listen():
        if message["type"] == "message":
            await render(message["data"])

Redis ne stocke rien. Si personne n’écoute, le message disparaît et PUBLISH renvoie simplement 0. C’est une livraison at-most-once : au mieux une fois, parfois jamais.

Cette absence de mémoire a aussi une conséquence moins visible. Un abonné trop lent voit ses messages s’accumuler dans le tampon de sortie de Redis. Au-delà de la limite client-output-buffer-limit pubsub (par défaut 32 Mo, ou 8 Mo pendant 60 secondes), Redis le déconnecte. Le client perd alors tout ce qui passe avant son retour, sans erreur côté producteur.

Pub/Sub reste excellent quand rater un message est sans gravité : un indicateur « en ligne », une invalidation de cache local avec expiration de secours, un signal « quelque chose a changé, viens relire ».

Redis Streams : un journal qu’on relit

Un stream est une structure persistante où l’on ajoute à la fin. Chaque entrée reçoit un identifiant croissant de la forme timestamp_ms-séquence :

message_id = await redis.xadd(
    "orders:events",
    {"type": "order.shipped", "payload": payload_json},
    maxlen=100_000,
    approximate=True,
)

La lecture ne supprime rien. Chaque lecteur tient son propre curseur et demande ce qui vient après :

async def follow(cursor: str) -> None:
    while True:
        batches = await redis.xread({"orders:events": cursor}, block=5_000, count=100)
        for _, messages in batches:
            for message_id, fields in messages:
                await render(fields)
                cursor = message_id

Reprenons le scénario de la coupure :

C’est ce mécanisme qui rend possible la reprise d’un flux Server-Sent Events : l’identifiant Redis sert d’id d’événement, et l’en-tête Last-Event-ID renvoyé par le navigateur devient le curseur. Le même principe sert au long polling, où le client relance sa requête depuis sa dernière position.

Attention toutefois à $, qui signifie « uniquement les nouvelles entrées ». Il est résolu à chaque appel : le repasser dans une boucle fait perdre ce qui est publié entre deux XREAD. Il sert au mieux pour le premier appel, ensuite on continue avec le dernier identifiant reçu.

Avec XREAD, tous les lecteurs voient tous les messages, et c’est à vous de conserver le curseur. C’est parfait pour de la diffusion rejouable, mais pas pour répartir du travail : deux workers lanceraient chacun le même traitement.

Les groupes de consommateurs : partager le travail

Pour répartir les messages entre plusieurs workers, Redis propose les consumer groups. Le groupe mémorise côté serveur où il en est. Dans un groupe, chaque message va à un seul consommateur :

                     ┌─ groupe billing ─┬─ worker-1 : A, C
orders:events ───────┤                  └─ worker-2 : B, D
(A, B, C, D)         └─ groupe emails ──── mailer-1 : A, B, C, D

Chaque groupe reçoit tout le flux, et les consommateurs d’un même groupe se le partagent. On obtient ainsi à la fois la diffusion entre services et la répartition de charge au sein d’un service.

from redis.exceptions import ResponseError

STREAM = "orders:events"
GROUP = "billing"


async def ensure_group() -> None:
    try:
        await redis.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
    except ResponseError as exc:
        if "BUSYGROUP" not in str(exc):
            raise


async def consume(consumer: str) -> None:
    while True:
        batches = await redis.xreadgroup(
            GROUP, consumer, {STREAM: ">"}, count=10, block=5_000
        )
        for _, messages in batches:
            for message_id, fields in messages:
                await handle(fields)
                await redis.xack(STREAM, GROUP, message_id)

L’identifiant spécial > demande les messages jamais livrés à ce groupe. Dès qu’un message est livré, il entre dans la Pending Entries List (PEL), la liste des messages reçus mais pas encore acquittés. XACK l’en retire.

L’ordre des lignes compte : on acquitte après le traitement. Si le worker s’arrête entre les deux, le message reste en attente et sera traité à nouveau.

Reprendre après un crash

Quand un worker meurt, ses messages restent dans la PEL à son nom. Personne ne les reprend automatiquement : si vous n’écrivez pas cette étape, ils y restent indéfiniment. C’est le rôle de XAUTOCLAIM, qui réattribue les messages inactifs depuis un certain temps :

import asyncio

MAX_DELIVERIES = 5


async def reclaim(consumer: str) -> None:
    cursor = "0-0"
    while True:
        cursor, messages, *_ = await redis.xautoclaim(
            STREAM, GROUP, consumer, min_idle_time=60_000, start_id=cursor, count=50
        )
        for message_id, fields in messages:
            pending = await redis.xpending_range(
                STREAM, GROUP, message_id, message_id, 1
            )
            if pending and pending[0]["times_delivered"] > MAX_DELIVERIES:
                await redis.xadd(f"{STREAM}:dead", fields)
                await redis.xack(STREAM, GROUP, message_id)
                continue
            await handle(fields)
            await redis.xack(STREAM, GROUP, message_id)
        if cursor == "0-0":
            await asyncio.sleep(30)

Deux choix méritent une explication.

min_idle_time doit être supérieur à la durée de traitement la plus longue. Sinon, un message encore en cours chez un worker lent sera repris par un autre, et traité deux fois en parallèle.

Le compteur times_delivered protège des messages empoisonnés. Un message qui fait planter chaque worker serait repris à l’infini. Au-delà du seuil, il part dans un stream dédié où il pourra être analysé. Sans cette limite, un seul message corrompu peut occuper vos workers en permanence.

Ce modèle est at-least-once : un message peut arriver deux fois, notamment après un crash entre handle et XACK. Le traitement doit donc supporter les doublons, avec un consommateur idempotent. À l’autre bout de la chaîne, le Transactional Outbox garantit que l’événement ajouté au stream correspond bien à un état réellement commité.

Ce que Streams ne garantit pas

La rétention ignore les acquittements

MAXLEN et XTRIM suppriment les entrées les plus anciennes sans vérifier si elles ont été acquittées. Un groupe en retard peut donc perdre des messages qu’il n’a jamais lus. Les messages déjà en attente restent référencés dans la PEL, mais leur contenu a disparu : depuis Redis 7.0, XAUTOCLAIM les retire de la liste et renvoie leurs identifiants dans son troisième élément, celui que l’exemple ignore.

Dimensionnez MAXLEN pour votre pire retard, et surveillez l’écart de chaque groupe avec XINFO GROUPS. Sans MAXLEN, le stream grossit jusqu’à ce que la mémoire manque.

La durabilité dépend de votre configuration Redis

Un stream vit en mémoire. Sa survie à un redémarrage dépend de la persistance choisie : avec appendfsync everysec, on peut perdre environ une seconde d’écritures. La réplication étant asynchrone, un failover peut aussi perdre les dernières entrées. Un XADD réussi ne prouve donc pas que le message survivra à une panne.

L’ordre n’est garanti qu’à la livraison

Le stream est ordonné, mais dès que plusieurs workers traitent en parallèle ou qu’un message est repris après un crash, l’ordre de traitement ne l’est plus. Si l’ordre métier compte, découpez par clé (un stream par commande ou par partition) ou rendez le traitement tolérant.

Pub/Sub ou Streams : comment choisir

Pub/Sub convient aux signaux éphémères, quand perdre un message ne casse rien.

XREAD convient à la diffusion rejouable vers des clients qui gardent leur position, comme un flux SSE.

Les groupes de consommateurs conviennent au traitement réparti, quand chaque message doit être pris en charge par un worker et survivre à son crash.

Et Kafka, Celery ou PostgreSQL ?

Redis Streams n’est pas un remplaçant universel.

Kafka devient pertinent quand la rétention se compte en semaines, que le volume dépasse ce que la mémoire peut porter, ou que vous avez besoin de partitions et d’un stockage disque répliqué. Si Redis est déjà dans votre stack et que le besoin tient en mémoire, Streams évite d’opérer un cluster de plus.

Celery répond à une autre question. Streams est un transport, alors que Celery fournit les retries avec délai, la planification, les résultats et la supervision des tâches. Pour « exécuter ce travail plus tard », Celery reste souvent le bon choix. Pour « réagir à ce flux d’événements », Streams est plus naturel.

PostgreSQL suffit parfois. Une table de travail lue avec select_for_update(skip_locked=True) partage aussi des tâches entre workers, avec un avantage difficile à égaler : l’ajout se fait dans la même transaction que vos données.

Conclusion

Le choix entre Pub/Sub et Streams n’est pas une question de performance, mais de garantie. Pub/Sub transmet à ceux qui sont là. XREAD laisse chaque lecteur rattraper son retard. Les groupes de consommateurs répartissent le travail et survivent à la perte d’un worker, à condition d’écrire la reprise, de limiter les messages empoisonnés et de tolérer les doublons.

Avant de choisir, posez la question de départ : si ce consommateur disparaît trente secondes, qu’est-ce qui doit se passer ? La réponse désigne presque toujours le bon outil.