What’s actually slowing this PC down?
Pick the symptom - the matching free tool is one click away.
Redis Streams permite añadir eventos ordenados, conservarlos para consultarlos y repartirlos entre consumidores de un grupo. Con XREADGROUP, XACK y la recuperación de mensajes pendientes, puedes diseñar procesamiento al menos una vez; no obtienes ejecución exactamente una vez de los efectos externos. WRedis ofrece una interfaz Python de conveniencia, pero la ficha de PyPI consultada no basta para atribuirle garantías de confirmación o recuperación.
Qué aporta Redis Streams frente a una entrega en vivo
Un stream es un registro de entradas. El productor las añade con XADD, Redis les asigna identificadores ordenados y luego se pueden consultar o reproducir por rangos con XRANGE. Esto resulta útil para telemetría, acciones de usuario, transacciones, mensajes entre servicios y notificaciones cuando importa conservar un historial.
As an Amazon Associate I earn from qualifying purchases.
Un grupo de consumidores reparte las entradas nuevas entre sus integrantes y mantiene su propio progreso. Otros grupos pueden leer el mismo stream de forma independiente. En cambio, Redis Pub/Sub es una distribución en vivo de mejor esfuerzo: un suscriptor que está desconectado no recibe los mensajes publicados durante su ausencia. Redis describe un grupo de consumidores como una estructura que sirve a varios consumidores y proporciona determinadas garantías; esas garantías no equivalen a ejecución exactamente una vez en toda la aplicación.
Cómo se coordinan la lectura y la confirmación
XGROUP CREATEcrea un grupo. El ID inicial$hace que empiece con las entradas nuevas;0-0permite comenzar desde el historial existente. Escoge según el comportamiento de bootstrap que necesites.XREADGROUPcon el marcador>entrega al grupo entradas nuevas que todavía no había repartido. Redis registra cada entrega no confirmada en la pending entries list (PEL).- El consumidor ejecuta el trabajo y, si termina correctamente, usa
XACKpara quitar la entrada de la PEL. XREADlee directamente el stream, pero no crea estado de grupo ni entradas confirmables en la PEL.
Implementación explícita con redis-py
Este ejemplo muestra el recorrido básico con una clave de stream llamada events. Usa un grupo que arranca desde el historial existente y confirma una entrada solo después de que el manejador termine. La creación del grupo debe gestionarse como inicialización: si ya existe, la aplicación debe tratar ese caso como un grupo previamente creado en vez de intentar recrearlo en cada arranque.
#1 Best Overall
import json
import redis
client = redis.Redis(host="localhost", port=6379, decode_responses=True)
stream = "events"
group = "processors"
consumer = "worker-1"
# Ejecutar como inicialización; elegir "$" si solo deben procesarse eventos nuevos.
client.xgroup_create(stream, group, id="0-0", mkstream=True)
# Productor: Redis genera el ID de la entrada.
event = {"event_id": "evt-1042", "type": "order.created", "payload": {"order_id": 1042}}
client.xadd(stream, {
"event_id": event["event_id"],
"type": event["type"],
"payload": json.dumps(event["payload"]),
})
# Consumidor: leer entradas nuevas para este grupo.
batches = client.xreadgroup(
groupname=group,
consumername=consumer,
streams={stream: ">"},
count=10,
block=5000,
)
for stream_name, entries in batches:
for entry_id, fields in entries:
payload = json.loads(fields["payload"])
handle_event(fields["event_id"], fields["type"], payload)
client.xack(stream_name, group, entry_id)handle_event representa el trabajo propio de la aplicación. Si falla y la ejecución no alcanza XACK, la entrada permanece pendiente. En una aplicación real, define también cómo se reportan errores y se reintentan o apartan los eventos que no pueden procesarse; no confirmes una entrada solo para despejar la PEL.
Redelivery, idempotencia y recuperación
La confirmación de Redis cubre el estado del stream, no una transacción atómica con una base de datos, un servicio de correo o un sistema de pagos externo. Si el manejador produce un efecto externo y el proceso cae antes de confirmar, otro consumidor puede ejecutar el mismo evento de nuevo. Diseña el manejador para tolerar esa repetición: por ejemplo, registra una clave de idempotencia derivada de event_id junto con el efecto, o deduplica en el sistema que lo recibe.
Para diagnosticar o recuperar entregas sin confirmar, inspecciona la PEL con XPENDING. XAUTOCLAIM permite reclamar entradas cuya inactividad supera un umbral y reasignarlas a otro consumidor. Por ejemplo, desde Redis CLI:
XAUTOCLAIM events processors worker-2 60000 0-0 COUNT 10En este ejemplo, 60000 es el umbral de inactividad en milisegundos, no un valor universal. Ajústalo por encima de la duración normal del trabajo para evitar que un consumidor activo pero lento pierda la entrada mientras aún la procesa. Si el umbral es demasiado corto, el trabajo puede ejecutarse simultáneamente en más de un consumidor.
Retención, reproducción y seguimiento operativo
La reproducción solo es posible mientras las entradas sigan retenidas. Redis permite limitar la longitud con XADD MAXLEN ~ o recortar por ID mínimo con XTRIM MINID ~. El recorte aproximado puede ser más eficiente que exigir un límite exacto, pero ambas estrategias pueden eliminar historial que un consumidor lento o una recuperación posterior necesitarían. Define la retención en función de la ventana de replay y recuperación que realmente requiera la aplicación.
XLENmuestra cuántas entradas hay en el stream.XINFO GROUPSpermite inspeccionar los grupos y su estado.XINFO CONSUMERSmuestra consumidores del grupo.XPENDINGayuda a distinguir entregas pendientes de entradas que aún no se han entregado.
Dónde encaja WRedis y qué comprobar
La ficha del paquete WRedis en PyPI anuncia RedisStreamManager y estos usos de interfaz:
from wredis.streams import RedisStreamManager
sm = RedisStreamManager(...)
sm.add_to_stream("events", {"type": "order.created"})
@sm.on_message("events", group_name="my_group", consumer_name="worker_1")
def process_message(...):
...La ficha también declara métodos exist, read_from_stream, wait y delete_stream. El ejemplo documenta la forma anunciada de la interfaz, no una validación independiente del comportamiento del decorador. La información consultada no establece las versiones compatibles de Redis o Python, si el decorador confirma automáticamente, cómo maneja reintentos o entradas atascadas, ni si garantiza un cierre ordenado. Antes de basar un sistema de producción en esas conductas, comprueba la documentación y versión publicadas y valida los casos de fallo que importan a tu aplicación.
Cuándo elegir Streams y qué depende de la versión
| Necesidad | Opción | Motivo |
|---|---|---|
| Historial, confirmaciones, recuperación y replay | Redis Streams con grupos | Mantiene progreso de grupo y entradas pendientes, además de un historial consultable. |
| Avisos transitorios para suscriptores conectados | Redis Pub/Sub | Distribuye en vivo, pero no retiene mensajes para suscriptores desconectados. |
| Una plataforma de streaming dedicada o requisitos operativos propios | Evaluar Kafka u otra plataforma según retención, escala y operación | La elección depende de la carga y del horizonte de replay; no hay un umbral universal de rendimiento establecido para una carga desconocida. |
Las funciones de Streams también varían según la versión del servidor. Redis documenta controles más detallados para coordinar varios grupos con XACKDEL, XDELEX, XADD y XTRIM desde Redis 8.2. La idempotencia de procesamiento de mensajes en Streams se documenta desde Redis 8.6; no presupongas que instalaciones anteriores incluyen esas funciones. El tutorial de Redis sobre ingesta con FastAPI pide Python 3.10 o posterior para esa demostración concreta, pero eso no establece una matriz de compatibilidad para WRedis.
Un patrón de aplicación para eventos malformados
Un tutorial de Redis publicado el 25 de marzo de 2026 muestra una aplicación FastAPI que añade eventos de telemetría a un stream, procesa un grupo, envía entradas malformadas a un stream de dead letter y registra métricas en Redis TimeSeries. Es un ejemplo de arquitectura, no una garantía de latencia para otras cargas. La separación en un stream de errores ayuda a conservar un registro inspeccionable de entradas que no pueden procesarse, mientras que las métricas permiten seguir el flujo sin confundirlo con la confirmación del mensaje.
Quick Recap
Product prices and availability are accurate as of the date/time indicated and are subject to change. Any price and availability information displayed on Amazon at the time of purchase will apply.
