Arquitectura basada en eventos: cómo sustituimos un sistema monolítico que costaba 50 000 dólares al mes

El problema: un monolito que no podía escalar

Antes: La cadena sincrónica del dolor

# La forma antigua: todo sincrónico, todo acoplado 
def process_order(request):    
 order = Order.objects.create(**request.data)    

 # Si CUALQUIERA de estas operaciones falla, el pedido falla    
 inventory_service.reserve(order)        # 200 ms    
 payment_service.charge(order)           # 800 ms    
 email_service.send_confirmation(order)  # 500 ms
    analytics_service.track(order)          # 300 ms
    warehouse_service.notify(order)         # 400 ms    

 # Total: tiempo de respuesta mínimo de 2200 ms
    # Si el correo electrónico no funciona: el pedido falla
    # Si el análisis es lento: el proceso de pago es lento
    return Response({'order_id': order.id})

La arquitectura: eventos y comandos

A continuación: Programación basada en eventos con Celery + Redis

# La nueva forma: sincrónico para la ruta crítica, eventos para el 
resto def process_order(request):    
 order = Order.objects.create(**request.data)    

 # Ruta crítica: debe completarse con éxito (sincrónico)    
 inventory_service.reserve(order)   # 200 ms    
 payment_service.charge(order)      # 800 ms    

 # Publicar evento: «disparar y olvidar» (asíncrono)
    publish_event('order.completed', {
        'order_id': order.id,
        'customer_email': order.customer.email,
        'total': str(order.total),
    })    

 # Total: 1000 ms de tiempo de respuesta
    # ¿El correo electrónico no funciona? El pedido sigue completándose. Se reintenta más tarde.
    return Response({'order_id': order.id})


# Gestores de eventos (trabajadores independientes)
@celery_app.task(bind=True, max_retries=5, default_retry_delay=60)
def handle_order_completed(self, event_data):
    """Cada gestor es independiente y se puede reintentar"""
    try:
        email_service.send_confirmation(event_data)
    except Exception as exc:
        self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

El patrón de bus de eventos

import redisimport
 
jsonfrom datetime import 

datetimeclass EventBus:
    def __init__(self):
        self.redis = redis.Redis(host='redis', port=6379, db=0)    

 def publish(self, event_type: str, data: dict):
        """Publicar evento en Redis Stream"""
        event = {
            'type': event_type,
            'data': json.dumps(data),
            'timestamp': datetime.utcnow().isoformat(),
        }
        self.redis.xadd(f'events:{event_type}', event)    

 def subscribe(self, event_type: str, group: str, consumer: str):
        """Leer eventos con un grupo de consumidores (cada evento se procesa una vez)"""
        try:
            self.redis.xgroup_create(
                f'events:{event_type}', group, id='0', mkstream=True
            )
        except redis.ResponseError:
            pass  # El grupo ya existe        

 while True:
            events = self.redis.xreadgroup(
                group, consumer,
                {f'events:{event_type}': '>'},
                count=10, block=5000
            )
            for stream, messages in events:
                for msg_id, fields in messages:
                    yield msg_id, json.loads(fields[b'data'])
                    self.redis.xack(stream, group, msg_id)

Resultados tras 3 meses

Cuándo NO utilizar una arquitectura basada en eventos

Empieza con un monolito. Extrae eventos cuando notes que te cuesta. La peor arquitectura es aquella que se construye para problemas que aún no tienes.

— alokknight Ingeniería