Redis Pub/Sub vs Streams: ¿cuál elegir en Python?

Redis Pub/Sub y Redis Streams parecen responder a la misma necesidad: un productor emite y unos consumidores reciben. Ambos se configuran en pocas líneas y, en un entorno de desarrollo, dan el mismo resultado.

La diferencia aparece con una sola pregunta: ¿qué ocurre si el consumidor no está presente cuando se envía el mensaje? Basta un despliegue, un corte de red o un worker que falla para que se plantee. La respuesta separa en realidad tres modelos: la difusión Pub/Sub, la lectura de un registro con XREAD y el reparto del trabajo con grupos de consumidores.

En resumen:

• Pub/Sub: para señales efímeras, un mensaje perdido no se recupera.
• XREAD: para una difusión reproducible, cada lector guarda su cursor.
• Grupos de consumidores: para repartir el trabajo entre workers, con recuperación tras un fallo.

Redis Pub/Sub: una difusión sin memoria

Con Pub/Sub, el productor publica en un canal y Redis transmite de inmediato el mensaje a los clientes suscritos en ese instante:

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 no almacena nada. Si nadie escucha, el mensaje desaparece y PUBLISH devuelve simplemente 0. Es una entrega at-most-once: una vez como mucho, a veces ninguna.

Esta falta de memoria tiene además una consecuencia menos visible. Los mensajes de un suscriptor lento se acumulan en el búfer de salida de Redis. Superado el límite client-output-buffer-limit pubsub (por defecto 32 MB, u 8 MB durante 60 segundos), Redis lo desconecta. El cliente pierde entonces todo lo enviado antes de volver, sin ningún error en el productor.

Pub/Sub sigue siendo excelente cuando perder un mensaje no tiene importancia: un indicador «en línea», una invalidación de caché local con expiración de respaldo, una señal de «algo ha cambiado, vuelve a leer».

Redis Streams: un registro que se puede releer

Un stream es una estructura persistente a la que se añade por el final. Cada entrada recibe un identificador creciente con la forma timestamp_ms-secuencia:

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

La lectura no elimina nada. Cada lector mantiene su propio cursor y pide lo que viene despué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

Volvamos al escenario del corte de red:

Este mecanismo es el que permite reanudar un flujo Server-Sent Events: el identificador de Redis sirve como id del evento, y la cabecera Last-Event-ID que envía el navegador se convierte en el cursor. La misma idea sirve para el long polling, donde el cliente repite la petición desde su última posición.

Cuidado, eso sí, con $, que significa «solo las entradas nuevas». Se resuelve en cada llamada: volver a pasarlo en un bucle hace perder lo que se publique entre dos XREAD. Como mucho sirve para la primera llamada; después se continúa con el último identificador recibido.

Con XREAD, todos los lectores ven todos los mensajes, y conservar el cursor es cosa tuya. Es perfecto para una difusión reproducible, pero no para repartir trabajo: dos workers ejecutarían cada uno la misma tarea.

Los grupos de consumidores: repartir el trabajo

Para repartir los mensajes entre varios workers, Redis ofrece los consumer groups. El grupo recuerda su posición en el servidor. Dentro de un grupo, cada mensaje va a un solo consumidor:

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

Cada grupo recibe todo el flujo, y los consumidores de un mismo grupo se lo reparten. Se obtiene así tanto la difusión entre servicios como el reparto de carga dentro de un servicio.

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)

El identificador especial > pide los mensajes nunca entregados a este grupo. En cuanto se entrega un mensaje, entra en la Pending Entries List (PEL), la lista de mensajes recibidos pero aún sin confirmar. XACK lo retira de ella.

El orden de estas líneas importa: se confirma después del procesamiento. Si el worker se detiene entre ambas, el mensaje sigue pendiente y se procesará de nuevo.

Recuperarse tras un fallo

Cuando un worker muere, sus mensajes permanecen en la PEL a su nombre. Nadie los recoge automáticamente: si no escribes este paso, se quedan ahí indefinidamente. Ese es el papel de XAUTOCLAIM, que reasigna los mensajes inactivos desde hace cierto tiempo:

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)

Dos decisiones merecen una explicación.

min_idle_time debe ser mayor que tu procesamiento más largo. De lo contrario, un mensaje que un worker lento aún está tratando será recogido por otro y procesado dos veces en paralelo.

El contador times_delivered protege de los mensajes envenenados. Un mensaje que hace fallar a todos los workers se reintentaría sin fin. Superado el umbral, pasa a un stream dedicado donde se puede analizar. Sin este límite, un solo mensaje corrupto puede tener ocupados a tus workers de forma permanente.

Este modelo es at-least-once: un mensaje puede llegar dos veces, sobre todo tras un fallo entre handle y XACK. El procesamiento debe por tanto tolerar duplicados, con un consumidor idempotente. En el otro extremo de la cadena, el Transactional Outbox garantiza que el evento añadido al stream corresponde a un estado realmente confirmado.

Lo que Streams no garantiza

La retención ignora las confirmaciones

MAXLEN y XTRIM eliminan las entradas más antiguas sin comprobar si se han confirmado. Un grupo retrasado puede por tanto perder mensajes que nunca leyó. Los mensajes ya pendientes siguen referenciados en la PEL, pero su contenido ha desaparecido: desde Redis 7.0, XAUTOCLAIM los retira de la lista y devuelve sus identificadores en su tercer elemento, el que el ejemplo ignora.

Dimensiona MAXLEN para tu peor retraso y vigila el desfase de cada grupo con XINFO GROUPS. Sin MAXLEN, el stream crece hasta que falta memoria.

La durabilidad depende de tu configuración de Redis

Un stream vive en memoria. Que sobreviva a un reinicio depende de la persistencia elegida: con appendfsync everysec, se puede perder alrededor de un segundo de escrituras. Como la replicación es asíncrona, un failover también puede perder las últimas entradas. Un XADD correcto no demuestra, por tanto, que el mensaje sobrevivirá a una avería.

El orden solo se garantiza en la entrega

El stream está ordenado, pero en cuanto varios workers procesan en paralelo o se reclama un mensaje tras un fallo, el orden de procesamiento deja de estarlo. Si el orden de negocio importa, divide por clave (un stream por pedido o por partición) o haz que el procesamiento sea tolerante.

Pub/Sub o Streams: cómo elegir

Pub/Sub conviene para señales efímeras, cuando perder un mensaje no rompe nada.

XREAD conviene para una difusión reproducible hacia clientes que guardan su posición, como un flujo SSE.

Los grupos de consumidores convienen para el procesamiento repartido, cuando cada mensaje debe ser atendido por un worker y sobrevivir a su fallo.

¿Y Kafka, Celery o PostgreSQL?

Redis Streams no es un sustituto universal.

Kafka resulta pertinente cuando la retención se mide en semanas, cuando el volumen supera lo que cabe en memoria o cuando necesitas particiones y almacenamiento replicado en disco. Si Redis ya está en tu stack y la necesidad cabe en memoria, Streams te evita operar un clúster más.

Celery responde a otra pregunta. Streams es un transporte, mientras que Celery aporta reintentos con retardo, planificación, resultados y supervisión de tareas. Para «ejecuta este trabajo más tarde», Celery suele seguir siendo la opción correcta. Para «reacciona a este flujo de eventos», Streams es más natural.

PostgreSQL a veces basta. Una tabla de trabajo leída con select_for_update(skip_locked=True) también reparte tareas entre workers, con una ventaja difícil de igualar: la inserción se hace en la misma transacción que tus datos.

Conclusión

Elegir entre Pub/Sub y Streams tiene menos que ver con el rendimiento que con las garantías. Pub/Sub entrega a quien esté presente. XREAD deja que cada lector se ponga al día. Los grupos de consumidores reparten el trabajo y sobreviven a la pérdida de un worker, siempre que escribas la recuperación, limites los mensajes envenenados y toleres los duplicados.

Antes de elegir, plantéate la pregunta inicial: si este consumidor desaparece treinta segundos, ¿qué debe ocurrir? La respuesta casi siempre señala la herramienta correcta.