DEV Community

William Rodriguez
William Rodriguez

Posted on

Ingestión confiable de eventos en Python: Redis Streams y Consumer Groups con WRedis

#streaming

Cuando tu aplicación supera las capacidades de Redis Pub/Sub tradicional, requieres persistencia en disco, control de offsets por consumidor y distribución horizontal de carga sin la complejidad de desplegar un broker pesado.

WRedis v1.0.0 LTS implementa abstracciones declarativas para Redis Streams (XADD, XREADGROUP) con decoradores automáticos para grupos de consumidores y parada limpia ante señales del sistema.

Este es el Día 03 de la serie técnica WRedis Open Source (MIT, Python 3.10+, 95%+ test coverage).

¿Por qué elegir Redis Streams frente a Pub/Sub?

  1. Historial Persistente: Los eventos se almacenan en el log de Redis y no se pierden si un worker se reinicia.
  2. Consumer Groups: Balanceo automático del procesamiento entre múltiples instancias sin duplicar mensajes.
  3. Pending Entries List (PEL): Recuperación garantizada de mensajes no confirmados tras caídas imprevistas.

Implementación en Producción: Streams con @on_message

1. Ingesta de Eventos Estructurados (add_to_stream) 

from wredis.streams import RedisStreamManager

stream_mgr = RedisStreamManager(host="localhost", port=6379)

# Enviar evento estructurado con serialización JSON automática
msg_id = stream_mgr.add_to_stream(
    key="eventos:transacciones",
    data={"tx_id": "TX-9941", "monto": 250.0, "moneda": "EUR"},
    ttl=86400  # Retención del stream por 24h
)
print(f"Evento registrado con ID: {msg_id}")
Enter fullscreen mode Exit fullscreen mode

2. Worker con Consumer Group Declarativo 

from wredis.streams import RedisStreamManager

stream_mgr = RedisStreamManager(host="localhost", verbose=False)

# Registro del worker con grupo de consumidores autogestionado
@stream_mgr.on_message(
    stream_name="eventos:transacciones",
    group_name="workers_liquidacion",
    consumer_name="nodo_primario"
)
def liquidar_transaccion(data):
    print(f"Liquidando TX: {data['tx_id']} por {data['monto']} {data['moneda']}")

# Iniciar bucle de consumo multihilo con parada limpia
stream_mgr.wait()
Enter fullscreen mode Exit fullscreen mode

Ventajas de WRedis en Producción

  • Grupos Sin Fricción: Crea los Consumer Groups automáticamente al iniciar sin arrojar excepciones si ya existen.
  • Hilos Gestionados: Los listeners corren en segundo plano con reconexión automática sin bloquear tu aplicación.
  • 95%+ Cobertura de Tests: Validado con 800+ pruebas unitarias, 38 de integración y 19 de estrés concurrente.

Instalación y Recursos 

pip install wredis
Enter fullscreen mode Exit fullscreen mode

Autor: William Steve Rodríguez Villamizar (Wisrovi)

Top comments (1)

Collapse
 
william_rodriguez_65a5898 profile image
William Rodriguez •

En sistemas distribuidos reactivos con Kafka, la frontera entre control de flujo y resiliencia ante caídas es sutil. El uso de patrones declarativos con commits atómicos permite que la lógica de reintento no degrade el rebalanceo de particiones.

¿Prefieren delegar el manejo de backpressure y offset commits a middleware declarativo o gestionarlo manualmente por handler?