Introducción a Colas de Trabajos con Redis y Patrones de Consumo

Concepto clave

Una cola de trabajos es un patrón de arquitectura que permite procesar tareas de forma asíncrona, separando la generación de trabajos de su ejecución. Imagina un restaurante: los clientes piden platos (generan trabajos), la cocina los prepara en orden (procesa los trabajos), y los camareros sirven cuando están listos (entregan resultados). En backend, esto evita bloquear respuestas HTTP mientras se realizan tareas pesadas como enviar emails, procesar imágenes o generar reportes.

Redis es ideal para colas por su velocidad en memoria y estructuras de datos como listas y sets ordenados. A diferencia de bases de datos tradicionales, Redis maneja millones de operaciones por segundo con latencia sub-milisegundo, crucial para sistemas de alta concurrencia. El patrón básico usa LPUSH para añadir trabajos y BRPOP para consumirlos, garantizando procesamiento ordenado y confiable.

Cómo funciona en la práctica

Veamos un ejemplo paso a paso para procesar notificaciones de usuario:

1. Productor: Un endpoint API recibe una solicitud para enviar email y añade un trabajo a la cola.

   import redis
   r = redis.Redis(host='localhost', port=6379)
   job_data = {'user_id': 123, 'email': '[email protected]', 'type': 'welcome'}
   r.lpush('queue:notifications', json.dumps(job_data))

2. Consumidor: Un worker separado escucha la cola y procesa trabajos.

   while True:
       job = r.brpop('queue:notifications', timeout=30)
       if job:
           data = json.loads(job[1])
           send_email(data['email'], data['type'])

3. Resultado: El usuario recibe el email sin que la API espere el envío, mejorando la latencia de respuesta.

Para colas prioritarias, usa sets ordenados con puntuación de prioridad:

Caso de estudio

Una plataforma de e-commerce procesa 10,000 pedidos por hora. Cada pedido requiere: validar stock, generar factura, y notificar al cliente. Usando Redis como cola:

• Arquitectura: Microservicio de pedidos añade trabajos a cola 'orders', workers separados procesan cada etapa.
• Implementación:

  # Productor
  order_data = {'order_id': 456, 'items': [1,2,3], 'user_email': '[email protected]'}
  r.lpush('queue:orders', json.dumps(order_data))
  
  # Consumidor (worker de facturación)
  job = r.brpop('queue:orders')
  data = json.loads(job[1])
  generate_invoice(data['order_id'])

• Resultados: Tiempo de respuesta API reducido de 2 segundos a 200ms, escalabilidad horizontal añadiendo más workers, y tolerancia a fallos con reintentos automáticos.

En producción, monitoriza la longitud de la cola con LLEN para detectar cuellos de botella y ajustar número de workers.

Errores comunes

1. Pérdida de trabajos por fallos del consumidor: Si un worker falla al procesar un trabajo, este se pierde. Solución: Usar BRPOPLPUSH para mover trabajos a una lista 'processing' y eliminarlos solo al confirmar éxito.
2. Colas bloqueantes sin timeout: BRPOP sin timeout puede consumir recursos indefinidamente. Siempre define un timeout razonable (ej. 30 segundos) para liberar conexiones.
3. Falta de priorización: Tratar todos los trabajos igual ralentiza críticos. Implementa colas múltiples (ej. 'high_priority', 'low_priority') o sets ordenados.
4. Serialización incorrecta Guardar objetos complejos sin JSON o Protocol Buffers causa errores. Usa siempre formatos estandarizados y valida datos.
5. Olvidar manejo de errores: No reintentar trabajos fallidos lleva a datos inconsistentes. Implementa cola de 'dead letters' para trabajos fallidos después de N intentos.

Checklist de dominio

• Puedo explicar la diferencia entre procesamiento síncrono y asíncrono con un ejemplo real.
• He implementado una cola simple en Redis usando LPUSH/BRPOP.
• Sé cómo priorizar trabajos usando sets ordenados o múltiples colas.
• Puedo diseñar un sistema de reintentos para trabajos fallidos.
• Entiendo cómo escalar horizontalmente añadiendo más consumidores.
• Sé monitorizar longitud de cola y tiempo de procesamiento con herramientas como Redis CLI.
• Puedo identificar cuándo usar Redis vs. sistemas de colas dedicados como RabbitMQ.

Implementa una cola de procesamiento de imágenes con Redis

En este ejercicio, crearás un sistema que procese miniaturas de imágenes de forma asíncrona usando Redis como cola.

1. Configura el entorno: Instala Redis localmente o usa un contenedor Docker. Asegúrate de tener Python 3 y la biblioteca redis-py.
2. Crea el productor: Escribe un script que simule subida de imágenes. Para cada 'upload', añade un trabajo a la cola 'image_queue' con datos JSON: {'image_id': 'uuid', 'path': '/uploads/img.jpg', 'size': 'thumbnail'}. Usa LPUSH.
3. Crea el consumidor: Implementa un worker que use BRPOP para obtener trabajos. Simula el procesamiento con: time.sleep(2) (representando generación de miniatura) y registra un mensaje de éxito.
4. Añade prioridad: Modifica el sistema para que imágenes de perfil (tipo 'avatar') tengan alta prioridad. Usa dos colas: 'high_priority' y 'low_priority', y un consumidor que verifique primero la cola alta.
5. Prueba el sistema: Ejecuta el productor para añadir 5 trabajos (2 de alta prioridad, 3 normales) y luego el consumidor. Verifica que los trabajos de alta prioridad se procesen primero.

• Usa json.dumps() para serializar datos antes de añadirlos a Redis.
• Para consumir de múltiples colas con prioridad, puedes usar BRPOP con múltiples nombres de lista.
• Considera usar threading en el consumidor para procesar múltiples trabajos concurrentemente.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.