# Redis Pub/Sub vs Streams : lequel choisir en Python ? > Redis Pub/Sub ou Streams ? Diffusion sans mémoire, journal rejouable, groupes de consommateurs et reprise après crash : les vraies garanties en Python. - URL: https://dev-flow.io/posts/redis-pub-sub-vs-streams/ - Date: 2026-11-04 - Author: Anthony - Language: fr - Categories: Architecture - Tags: Python, Architecture, Redis, Messaging - Translations: [en](https://dev-flow.io/en/posts/redis-pub-sub-vs-streams/index.md), [es](https://dev-flow.io/es/posts/redis-pub-sub-vs-streams/index.md) --- 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 : ```python 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` : ```python 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 : ```python 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 : | Temps | Événement | Pub/Sub | Streams avec `XREAD` | |---|---|---|---| | t0 | Le client reçoit A | Reçu | Reçu, curseur sur A | | t1 | Coupure réseau | | | | t2 | B est publié | Perdu | Conservé | | t3 | C est publié | Perdu | Conservé | | t4 | Le client revient | Rien à rejouer | `XREAD` depuis A renvoie B et C | C'est ce mécanisme qui rend possible la reprise d'un flux [Server-Sent Events](/posts/server-sent-events-vs-websocket/) : 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](/posts/polling-long-polling-sse/), 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 : ```text ┌─ 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. ```python 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 : ```python 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](/posts/pattern-inbox-consumer-idempotent/). À l'autre bout de la chaîne, le [Transactional Outbox](/posts/pattern-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 | Besoin | Pub/Sub | `XREAD` | Groupes de consommateurs | |---|---|---|---| | Message conservé | Non | Oui, jusqu'à éviction | Oui, jusqu'à éviction | | Livraison | At-most-once | Relecture depuis un curseur | At-least-once | | Qui reçoit | Tous les abonnés connectés | Tous les lecteurs | Un consommateur par groupe | | Curseur | Aucun | Côté client | Côté Redis | | Reprise après crash | Aucune | Avec le curseur | `XAUTOCLAIM` | | Coût mémoire | Quasi nul | À borner | À borner, PEL à surveiller | **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)`](/posts/django-select-for-update/) 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.