Ir al contenido
wannacode

Eventos y colas

En resumen: Un evento dice «esto pasó» y va a una cola de la que cada suscriptor lo toma por su cuenta; la entrega es al menos una vez, así que el consumidor tiene que ser idempotente.

Problema

El correo lento frena el pago

Cuando un cliente paga, la tienda hace cuatro cosas: cobra la tarjeta, reserva el stock en el almacén, envía un correo de confirmación y anota el pago en la contabilidad. El cobro las llama una tras otra:

card = Service("tarjeta", 2)
warehouse = Service("almacén", 3)
mailer = Service("correo", 25)  # SMTP lento: 25 ticks
ledger = Service("contabilidad", 4)




def pay(order, total):
    start = Service.clock
    card.call(f"cobra {total} €")
    warehouse.call(f"reserva stock para {order}")
    mailer.call(f"confirmación del pedido {order}")
    ledger.call(f"{total} € para {order}")
    waited = Service.clock - start
    print(f"✓ {order}: el cliente espera {waited} ticks")

Cada «servicio» aquí es un objeto con un contador: cada llamada suma sus ticks. Los ticks son el tiempo del programa, en vez de segundos. El correo pasa por un servidor de correo lento y cuesta 25 ticks:

class Service:
    clock = 0  # los ticks que pasan


    def __init__(self, name, cost, failures=0):
        self.name = name
        self.cost = cost
        self.failures = failures  # escenario grabado: cuántas veces seguidas falla


    def call(self, what):
        Service.clock += self.cost
        if self.failures > 0:
            self.failures -= 1
            raise RuntimeError(f"{self.name} no responde")
        print(f"{self.name}: {what}")

El primer pago sale bien. Luego el correo falla una vez:

pay(1001, 120)
mailer.failures = 1  # el correo falla una vez
try:
    pay(1002, 80)
except RuntimeError as e:
    print(f"✗ 1002: {e}")
Salida
tarjeta: cobra 120 €
almacén: reserva stock para 1001
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
✓ 1001: el cliente espera 34 ticks
tarjeta: cobra 80 €
almacén: reserva stock para 1002
✗ 1002: correo no responde

El cliente espera 34 ticks, 25 de ellos por un correo que podría llegar perfectamente un minuto después. Y con el pedido 1002 la tarjeta está cobrada y el stock reservado, pero el cliente ve un error, porque el servidor de correo no responde. Pulsa «Pagar» otra vez.

Hay un tercer dolor. El cobro conoce a todos los que reaccionan. Mañana marketing quiere puntos de fidelidad y la app móvil quiere un aviso. Cada línea nueva es un cambio en el código del cobro.

La idea

Un evento: esto ya pasó

El cobro no debería decirle al almacén «reserva», al correo «envía» y a la contabilidad «anota». Basta con anunciar lo que pasó: «el pedido 1001 está pagado». Esto es un evento de dominio: un hecho en el lenguaje de la tienda, en pasado, con un nombre como OrderPaid. Lleva su propio id y los datos que necesitan los demás: el número de pedido, el total, la cantidad. Lo que ya pasó no cambia.

El evento va a una cola de mensajes. El programa que guarda las colas es un broker. Quien quiera enterarse de los pagos se suscribe. El broker guarda una cola aparte para cada suscriptor y pone una copia del evento en cada una. Esto es publish/subscribe: el cobro publica, los suscriptores reciben, y ninguno de los lados sabe del otro.

Ya viste la idea en Observer: quien quiere enterarse se suscribe. La diferencia es la cola en medio. Observer llama a los suscriptores enseguida, en la misma llamada, y espera a cada uno. Aquí el cobro deja el evento en la cola y le responde al cliente. Los suscriptores lo toman cuando pueden, como los empacadores en productor y consumidor.

El cobro y el almacén ya pueden ser módulos separados, incluso servicios separados, como en la lección anterior: hablan solo mediante eventos.

Míralo

El broker en memoria

Un broker real es un programa aparte (Kafka, RabbitMQ). Aquí es una clase en memoria con un mapa de colas, para que veas el mecanismo. El evento:

@dataclass(frozen=True)
class OrderPaid:
    id: str
    order: int
    total: int
    qty: int
En Python el evento es una @dataclass(frozen=True): intentar cambiar un campo lanza un error.

El broker. Un suscriptor da un nombre y una función que procesa el evento:

class Broker:
    def __init__(self, network):
        self.queues = {}  # suscriptor → su propia cola
        self.handlers = {}
        self.network = network


    def subscribe(self, name, handler):
        self.queues[name] = deque()
        self.handlers[name] = handler


    def publish(self, event):
        for queue in self.queues.values():
            queue.append(event)  # una copia para cada suscriptor


    # una ronda: cada suscriptor recibe el primero de su cola
    def deliver(self):
        any_left = False
        for name, queue in self.queues.items():
            if not queue:
                continue
            any_left = True
            try:
                self.handlers[name](queue[0])
            except RuntimeError as e:
                print(f"  ✗ {e}: sigue en la cola")
                continue
            # el ack se pierde: el evento se queda y llegará otra vez
            if self.network.ack_lost(name):
                continue
            queue.popleft()  # ack: sácalo de la cola
        return any_left

Lo importante está en deliver. El suscriptor recibe el primer evento de su cola, pero el evento sale de la cola solo cuando el suscriptor termina sin error. La confirmación «ya terminé» se llama ack (acknowledgement). Si el suscriptor lanza una excepción, no hay ack y el evento se queda para la ronda siguiente. Un ack perdido viene de la clase Network: un escenario grabado de a quién se le pierde el ack. Aquí no pierde nada; en la tarjeta siguiente sí.

class Network:
    def __init__(self, *lost):
        self.lost = list(lost)  # escenario grabado: de quién se pierde el ack


    def ack_lost(self, name):
        if name not in self.lost:
            return False
        self.lost.remove(name)
        print(f"  ✗ ack de {name} se pierde")
        return True

El cobro ahora solo cobra la tarjeta y publica:

card = Service("tarjeta", 2)




def pay(order, total, qty):
    start = Service.clock
    card.call(f"cobra {total} €")
    broker.publish(OrderPaid(f"evt-{order}", order, total, qty))
    waited = Service.clock - start
    print(f"✓ {order}: el cliente espera {waited} ticks")

Dos pedidos, y el correo falla una vez, como en el problema:

mailer = Service("correo", 25, 1)
ledger = Service("contabilidad", 4)
broker = Broker(Network())
warehouse = Warehouse()
broker.subscribe("almacén", warehouse.on)
broker.subscribe("correo", lambda e: mailer.call(f"confirmación del pedido {e.order}"))
broker.subscribe("contabilidad", lambda e: ledger.call(f"{e.total} € para {e.order}"))
pay(1001, 120, 2)
pay(1002, 80, 1)
while broker.deliver():
    pass
Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
tarjeta: cobra 80 €
✓ 1002: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
  ✗ correo no responde: sigue en la cola
contabilidad: 120 € para 1001
almacén: 1 unidades para 1002, quedan 7
correo: confirmación del pedido 1001
contabilidad: 80 € para 1002
correo: confirmación del pedido 1002

El cliente espera 2 ticks en los dos pedidos. El correo sigue lento y sigue fallando, pero ahora es problema suyo: el evento se queda en su cola y vuelve en la ronda siguiente. El almacén y la contabilidad ni se enteran.

Fíjate también en el orden. La confirmación del 1001 sale después de que el almacén procesó el 1002. Cada cola avanza a su ritmo. Por eso un suscriptor no puede contar con que otro ya hizo su parte.

La idea

Al menos una vez significa a veces dos

El broker quita el evento de la cola solo después del ack. ¿Qué pasa si el almacén saca el stock, pero el ack se pierde en la red? El broker no distingue «el almacén falló antes de empezar» de «el almacén terminó, pero no pudo decirlo». Para no perder un pedido, entrega el evento otra vez.

El escenario pierde el ack del almacén una vez. El almacén es ingenuo: cada evento saca stock:

class Warehouse:
    def __init__(self):
        self.stock = 10


    def on(self, e):
        self.stock -= e.qty
        print(f"almacén: {e.qty} unidades para {e.order}, quedan {self.stock}")
mailer = Service("correo", 25, 0)
ledger = Service("contabilidad", 4)
broker = Broker(Network("almacén"))
warehouse = Warehouse()
broker.subscribe("almacén", warehouse.on)
broker.subscribe("correo", lambda e: mailer.call(f"confirmación del pedido {e.order}"))
broker.subscribe("contabilidad", lambda e: ledger.call(f"{e.total} € para {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
  ✗ ack de almacén se pierde
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
almacén: 2 unidades para 1001, quedan 6

Un pedido de 2 unidades, y al almacén le faltan 4. El stock está mal, y la tienda rechazará el pedido siguiente aunque la mercancía esté en el estante.

Un broker puede prometer una de dos cosas:

  • como máximo una vez: quita el evento antes de que el suscriptor trabaje. Sin duplicados, pero si el suscriptor falla, el evento se pierde;
  • al menos una vez: lo quita después del ack. No se pierde nada, pero a veces llega dos veces.

«Exactamente una vez» a través de una red es algo que un broker no puede prometer por sí solo. Por eso casi todos los sistemas eligen «al menos una vez», y el receptor atrapa el duplicado.

Míralo

Un consumidor idempotente

Un consumidor para quien recibir el mismo evento por segunda vez no cambia nada es idempotente (idempotent consumer). El almacén recuerda los ids de los eventos que ya procesó y descarta el duplicado:

class Warehouse:
    def __init__(self):
        self.stock = 10
        # los ids de los eventos ya procesados
        self.seen = set()


    def on(self, e):
        if e.id in self.seen:
            print(f"almacén: {e.id} es un duplicado")
            return
        self.stock -= e.qty
        self.seen.add(e.id)
        print(f"almacén: {e.qty} unidades para {e.order}, quedan {self.stock}")

def publish(self, event):
    for queue in self.queues.values():
        queue.append(event)  # una copia para cada suscriptor


# una ronda: cada suscriptor recibe el primero de su cola
def deliver(self):
    any_left = False
    for name, queue in self.queues.items():
        if not queue:
            continue
        any_left = True
        try:
            self.handlers[name](queue[0])
        except RuntimeError as e:
            print(f"  ✗ {e}: sigue en la cola")
            continue
        # el ack se pierde: el evento se queda y llegará otra vez
        if self.network.ack_lost(name):
            continue
        queue.popleft()  # ack: sácalo de la cola
    return any_left
Paso a paso, en palabras
  1. El cobro carga la tarjeta y publica OrderPaid evt-1001: «el pedido 1001 está pagado». El cliente espera solo 2 ticks.
  2. El broker pone una copia en cada cola: almacén, correo y contabilidad. El cobro ya ha respondido.
  3. En cada ronda cada suscriptor recibe el primer evento de su propia cola. El evento sigue en la cola hasta el ack.
  4. El almacén saca 2 unidades (quedan 8) y anota evt-1001 en seen. El correo y la contabilidad hacen lo suyo.
  5. Los acks del correo y de la contabilidad llegan, y el evento sale de sus colas. El ack del almacén se pierde por el camino.
  6. El broker no sabe que el almacén ya hizo el trabajo y entrega evt-1001 otra vez. Esto es «al menos una vez».
  7. evt-1001 ya está en seen, así que el almacén lo descarta. El stock se queda en 8, no en 6.
  8. Esta vez el ack llega y la cola queda vacía. Entregado al menos una vez, procesado exactamente una vez.
mailer = Service("correo", 25, 0)
ledger = Service("contabilidad", 4)
broker = Broker(Network("almacén"))
warehouse = Warehouse()
broker.subscribe("almacén", warehouse.on)
broker.subscribe("correo", lambda e: mailer.call(f"confirmación del pedido {e.order}"))
broker.subscribe("contabilidad", lambda e: ledger.call(f"{e.total} € para {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
  ✗ ack de almacén se pierde
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
almacén: evt-1001 es un duplicado

El broker sigue entregando dos veces, pero el almacén saca el stock una sola vez. El id se escribe después del trabajo: si el trabajo falla a medias, el evento vuelve y se procesa.

Algunas operaciones son idempotentes por sí solas: «el estado de 1001 es pagado» es lo mismo la segunda vez. «Stock menos 2» no. Para esas hace falta un id. En un sistema real seen es una tabla en la misma base de datos que el stock, y la resta y el id se escriben en una sola transacción. Si no, una caída entre las dos vuelve a dar un resultado incorrecto. Qué pasa cuando la escritura en la base de datos y el envío del evento están en sistemas distintos se cuenta en la lección siguiente.

Pruébalo

Predice la salida

El almacén es idempotente. Esta vez el escenario pierde el ack del correo. ¿Qué imprime el programa?

mailer = Service("correo", 25, 0)
ledger = Service("contabilidad", 4)
broker = Broker(Network("correo"))
warehouse = Warehouse()
broker.subscribe("almacén", warehouse.on)
broker.subscribe("correo", lambda e: mailer.call(f"confirmación del pedido {e.order}"))
broker.subscribe("contabilidad", lambda e: ledger.call(f"{e.total} € para {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Respuesta

Pruébalo

Encuentra el error

Un colega reescribe el almacén, y la prueba con el ack perdido sigue mostrando el stock sacado dos veces. ¿Qué línea tiene el error?

Elige la línea con el error.

La idea

Cuándo los eventos hacen daño

Los eventos no son gratis. No los uses cuando:

  • necesitas la respuesta ahora mismo. «¿Hay stock?» antes del pago es una pregunta, no un evento. Si no puedes seguir sin la respuesta, llama al otro módulo;
  • el sistema es pequeño y está en un solo proceso. Basta una llamada u Observer. Un broker es un programa más que hay que mantener en marcha, vigilar y actualizar.

Y conoce el precio cuando sí los uses:

  • el flujo ya no está en un solo lugar. Lo que pasa después del pago no se lee en una sola función; lo reúnes desde los suscriptores. Por eso cada evento lleva un id que luego puedes buscar en los logs;
  • los demás se enteran después. El cliente paga y abre enseguida «Mis pedidos», y la factura todavía no está. Esto se llama consistencia eventual. La pantalla debería decirlo con honestidad: «la factura llega en un minuto»;
  • no se promete un orden. Lo viste en el programa: las colas avanzan a su ritmo. Si el orden importa (pagado, luego cancelado), los eventos de un pedido deben pasar por una sola cola y un solo consumidor;
  • los duplicados son cosa tuya. Todos los suscriptores tienen que ser idempotentes.

Cómo se opera y se vigila un broker real en producción (un clúster, discos, monitoreo) es tema del curso de DevOps ZeroToOps.

Más precisamente, para curiosos

Brokers reales. Kafka guarda los eventos como un log que cada suscriptor lee desde su propia posición. Allí el ack es la posición recordada (el offset). RabbitMQ tiene colas y un ack para cada mensaje, como en la lección. También existen Redis Streams, Amazon SQS y SNS, Azure Service Bus y Google Pub/Sub. En Spring un suscriptor es un método con @KafkaListener o @RabbitListener. Los proyectos de ASP.NET Core suelen usar MassTransit. Python tiene clientes como confluent-kafka y aio-pika, y Celery es una cola de tareas. Node.js tiene kafkajs y amqplib, y NestJS tiene un módulo de mensajería.

La dead letter queue. Un evento que falla siempre (datos rotos) no debe dar vueltas en la cola para siempre. Tras unos intentos el broker lo pasa a una cola aparte donde lo mira una persona.

Exactamente una vez. Kafka tiene un modo «exactly once», pero vale dentro de Kafka: leer, procesar, volver a escribir en Kafka. En cuanto el efecto está fuera (un correo, un pago, una base de datos), el duplicado sigue siendo cosa tuya.

Evento u orden. OrderPaid es un evento: muchos suscriptores, ninguno obligado. ReserveStock es una orden: un receptor que debe cumplirla. Ambos pueden pasar por una cola, pero los nombres expresan intenciones distintas.

Cómo es en los lenguajes. En Python un suscriptor es cualquier cosa que se pueda llamar: un método de un objeto (warehouse.on) o una lambda. No hay interfaz: el broker llama a lo que le des. deque quita rápido por delante, mientras que list.pop(0) mueve todos los elementos.

Ejercicio

Un aviso en la app

Resuelve la tarea en tu editor.

La app móvil quiere un aviso de «pedido pagado». Añade un cuarto suscriptor, «app», sin tocar el código del cobro. Tiene que ser idempotente, como el almacén. El escenario pierde su ack una vez:

Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
app: aviso del pedido 1001
  ✗ ack de app se pierde
app: evt-1001 es un duplicado

Más tareas

  • Un pedido cancelado. Añade un evento OrderCancelled. El almacén devuelve el stock, pero solo si procesó OrderPaid del mismo pedido. ¿Qué pasa si la cancelación llega antes que el pago?
  • Una dead letter queue. Si un suscriptor falla tres veces seguidas con el mismo evento, el broker lo pasa a una lista dead y sigue con el siguiente.
  • Puntos de fidelidad. Añade un suscriptor que da 1 punto por cada 10 € del total. ¿Qué tiene que recordar para no dar los puntos dos veces?

Test

Ponte a prueba

1. ¿Qué nombre es un evento de dominio y no una orden?
2. El ack del almacén se pierde por el camino. ¿Qué hace el broker?
3. ¿Por qué el almacén escribe el id en seen DESPUÉS de sacar el stock y no antes?
4. ¿Cuándo es mejor llamar directamente al otro módulo en vez de publicar un evento?

Creado por Ivo Gergov como parte de la campaña de Evo Tech por una educación accesible en programación ·