Redis Pub/Sub vs Streams: Which to Choose in Python?

Redis Pub/Sub and Redis Streams seem to meet the same need: a producer emits, consumers receive. Both take a few lines to set up, and in a development environment they produce the same result.

The difference shows up with a single question: what happens if the consumer isn’t there when the message is sent? A deployment, a network drop or a crashed worker is enough to raise it. The answer actually separates three models: Pub/Sub broadcast, reading a log with XREAD, and sharing work with consumer groups.

In short:

• Pub/Sub: for ephemeral signals, a missed message is lost.
• XREAD: for replayable broadcast, each reader keeps its own cursor.
• Consumer groups: to share work between workers, with crash recovery.

Redis Pub/Sub: a broadcast with no memory

With Pub/Sub, the producer publishes on a channel and Redis immediately forwards the message to the clients subscribed at that moment:

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 stores nothing. If nobody is listening, the message vanishes and PUBLISH simply returns 0. This is at-most-once delivery: once at best, sometimes never.

This lack of memory has a less visible consequence too. A slow subscriber’s messages pile up in the Redis output buffer. Beyond the client-output-buffer-limit pubsub limit (by default 32 MB, or 8 MB for 60 seconds), Redis disconnects it. The client then loses everything sent before it comes back, with no error on the producer side.

Pub/Sub remains excellent when missing a message is harmless: an “online” indicator, a local cache invalidation backed by an expiry, a “something changed, come and read again” signal.

Redis Streams: a log you can reread

A stream is a persistent structure you append to. Each entry receives an increasing ID of the form timestamp_ms-sequence:

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

Reading deletes nothing. Each reader keeps its own cursor and asks for what comes next:

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

Let’s go back to the network drop scenario:

This mechanism is what makes it possible to resume a Server-Sent Events stream: the Redis ID serves as the event id, and the Last-Event-ID header sent back by the browser becomes the cursor. The same idea powers long polling, where the client retries from its last position.

Be careful with $, though, which means “new entries only”. It is resolved on each call: passing it again in a loop loses whatever is published between two XREAD calls. At most, use it for the first call, then continue with the last ID received.

With XREAD, every reader sees every message, and keeping the cursor is up to you. That is perfect for replayable broadcast, but not for distributing work: two workers would each run the same job.

Consumer groups: sharing the work

To distribute messages across several workers, Redis offers consumer groups. The group remembers its position on the server side. Within a group, each message goes to a single consumer:

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

Each group receives the whole stream, and the consumers of a group share it. You get both broadcast between services and load balancing within a 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)

The special > ID asks for messages never delivered to this group. As soon as a message is delivered, it enters the Pending Entries List (PEL), the list of messages received but not yet acknowledged. XACK removes it.

The order of these lines matters: you acknowledge after processing. If the worker stops between the two, the message stays pending and will be processed again.

Recovering after a crash

When a worker dies, its messages stay in the PEL under its name. Nobody picks them up automatically: if you don’t write this step, they stay there indefinitely. That is the job of XAUTOCLAIM, which reassigns messages that have been idle for some time:

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)

Two choices deserve an explanation.

min_idle_time must be longer than your longest processing time. Otherwise, a message still being handled by a slow worker will be picked up by another one and processed twice in parallel.

The times_delivered counter protects against poison messages. A message that crashes every worker would be retried forever. Beyond the threshold, it goes to a dedicated stream where it can be analyzed. Without this limit, a single corrupted message can keep your workers busy permanently.

This model is at-least-once: a message can arrive twice, notably after a crash between handle and XACK. Processing must therefore tolerate duplicates, with an idempotent consumer. At the other end of the chain, the Transactional Outbox guarantees that the event added to the stream matches a state that was really committed.

What Streams does not guarantee

Retention ignores acknowledgements

MAXLEN and XTRIM delete the oldest entries without checking whether they were acknowledged. A group that lags behind can therefore lose messages it never read. Messages already pending stay referenced in the PEL, but their content is gone: since Redis 7.0, XAUTOCLAIM removes them from the list and returns their IDs in its third element, the one the example ignores.

Size MAXLEN for your worst delay, and monitor each group’s lag with XINFO GROUPS. Without MAXLEN, the stream grows until memory runs out.

Durability depends on your Redis configuration

A stream lives in memory. Whether it survives a restart depends on the persistence you chose: with appendfsync everysec, you can lose about one second of writes. Since replication is asynchronous, a failover can also lose the latest entries. A successful XADD therefore doesn’t prove the message will survive a failure.

Order is only guaranteed at delivery

The stream is ordered, but as soon as several workers process in parallel or a message is reclaimed after a crash, processing order no longer is. If business order matters, split by key (one stream per order or per partition) or make processing tolerant.

Pub/Sub or Streams: how to choose

Pub/Sub suits ephemeral signals, when losing a message breaks nothing.

XREAD suits replayable broadcast to clients that keep their position, such as an SSE stream.

Consumer groups suit distributed processing, when each message must be handled by a worker and survive its crash.

What about Kafka, Celery or PostgreSQL?

Redis Streams is not a universal replacement.

Kafka becomes relevant when retention is measured in weeks, when volume exceeds what memory can hold, or when you need partitions and replicated disk storage. If Redis is already in your stack and the need fits in memory, Streams saves you from operating another cluster.

Celery answers a different question. Streams is a transport, whereas Celery provides delayed retries, scheduling, results and task supervision. For “run this job later”, Celery is often still the right choice. For “react to this stream of events”, Streams is more natural.

PostgreSQL is sometimes enough. A work table read with select_for_update(skip_locked=True) also shares jobs between workers, with an advantage that is hard to match: the insert happens in the same transaction as your data.

Conclusion

Choosing between Pub/Sub and Streams is less about performance than about guarantees. Pub/Sub delivers to whoever is there. XREAD lets each reader catch up. Consumer groups distribute work and survive the loss of a worker, provided you write the recovery, limit poison messages and tolerate duplicates.

Before choosing, ask the opening question: if this consumer disappears for thirty seconds, what should happen? The answer almost always points to the right tool.