Diseñar Sistemas Event-Driven con Event Buses y Brokers

Visión general

Las llamadas síncronas entre servicios crean acoplamiento fuerte. El servicio que llama tiene que saber dónde vive el otro, esperar una respuesta y gestionar los fallos él mismo. Cuando el servicio llamado va lento o está caído, el que llama sufre con él. A medida que los sistemas crecen, esta red de dependencias directas se convierte en el verdadero problema: un cambio en un servicio se propaga a todos los que lo invocan.

La arquitectura event-driven le da la vuelta a esta relación. En lugar de llamarse directamente, los servicios dejan eventos en un broker y siguen a lo suyo. El evento OrderPlaced sale una sola vez: inventario se suscribe y descuenta stock, facturación emite la factura y envíos prepara la etiqueta. Cada uno trabaja a su propio ritmo — si facturación va lenta, pedidos y envíos siguen funcionando. Las secciones siguientes cubren la parte práctica: patrones de eventos, cómo elegir broker e implementaciones funcionales en Kafka, RabbitMQ y AWS EventBridge.

Cuándo usarlo

Usa esta receta cuando:

• Varios servicios deben reaccionar al mismo evento de negocio
• Las cargas son irregulares y necesitas buffering para suavizar picos de tráfico
• Los servicios tienen requisitos de disponibilidad distintos y no pueden bloquearse entre sí
• Necesitas un audit trail donde cada cambio de estado quede registrado
• Implementas event sourcing para consultas temporales y reconstrucción de estado — consulta CQRS Pattern para separación lectura/escritura

Sáltalo cuando todo cabe en una sola unidad desplegable — un broker añade carga operativa que un servicio único nunca recupera. También es la herramienta equivocada cuando una petición necesita una respuesta inmediata y consistente (autorización de pago, un check de stock en el checkout): los eventos se resuelven eventualmente, no de forma síncrona.

Solución

Publicando Eventos (Python / Kafka)

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',
    retries=3,
)

def place_order(order_data):
    # Guardar el pedido en base de datos
    order = save_order(order_data)

    # Publicar el evento
    event = {
        'type': 'OrderPlaced',
        'aggregate_id': order.id,
        'payload': {
            'customer_id': order.customer_id,
            'items': [item.to_dict() for item in order.items],
            'total': order.total(),
        },
        'occurred_at': order.created_at.isoformat(),
    }

    producer.send('orders', key=order.id.encode(), value=event)
    producer.flush()

    return order

Consumiendo Eventos (Node.js / RabbitMQ)

const amqp = require('amqplib');

async function startInventoryConsumer() {
  const connection = await amqp.connect('amqp://rabbitmq');
  const channel = await connection.createChannel();

  const queue = 'inventory_updates';
  await channel.assertQueue(queue, { durable: true });
  await channel.bindQueue(queue, 'orders', 'OrderPlaced');

  channel.consume(queue, async (msg) => {
    if (msg !== null) {
      const event = JSON.parse(msg.content.toString());
      console.log(`Procesando ${event.type} para el pedido ${event.aggregate_id}`);

      try {
        await reserveInventory(event.payload.items);
        channel.ack(msg); // Confirmar procesamiento
      } catch (error) {
        channel.nack(msg, false, false); // A dead-letter, no reencolar
      }
    }
  });
}

startInventoryConsumer();

AWS EventBridge Event Bus (Terraform)

resource "aws_cloudwatch_event_bus" "main" {
  name = "stackpractices-events"
}

resource "aws_cloudwatch_event_rule" "order_placed" {
  name        = "order-placed-rule"
  event_bus_name = aws_cloudwatch_event_bus.main.name

  event_pattern = jsonencode({
    source      = ["order-service"]
    detail-type = ["OrderPlaced"]
  })
}

resource "aws_cloudwatch_event_target" "inventory_target" {
  rule           = aws_cloudwatch_event_rule.order_placed.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  arn            = aws_sqs_queue.inventory_queue.arn
}

resource "aws_cloudwatch_event_target" "billing_target" {
  rule           = aws_cloudwatch_event_rule.order_placed.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  arn            = aws_lambda_function.billing_processor.arn
}

Event Sourcing con Kafka Streams (Java)

public class OrderEventStore {

    private final KafkaStreams streams;

    public OrderEventStore() {
        StreamsBuilder builder = new StreamsBuilder();

        // Event store: agregar eventos por ID de pedido
        KStream<String, OrderEvent> eventStream = builder.stream(
            "orders",
            Consumed.with(Serdes.String(), new OrderEventSerde())
        );

        // Materializar el estado actual desde el historial de eventos
        KTable<String, OrderState> orderState = eventStream
            .groupByKey()
            .aggregate(
                OrderState::new,
                (key, event, state) -> state.apply(event),
                Materialized.as("order-state-store")
            );

        // Proyectar a un tópico de read model
        orderState.toStream().to("order-read-model",
            Produced.with(Serdes.String(), new OrderStateSerde()));

        streams = new KafkaStreams(builder.build(), getStreamsConfig());
    }

    public void start() {
        streams.start();
    }

    public OrderState getOrder(String orderId) {
        ReadOnlyKeyValueStore<String, OrderState> store =
            streams.store(StoreQueryParameters.fromNameAndType(
                "order-state-store",
                QueryableStoreTypes.keyValueStore()
            ));
        return store.get(orderId);
    }
}

// Aplicar eventos para reconstruir el estado
class OrderState {
    private String status;
    private BigDecimal total;
    private List<String> items = new ArrayList<>();

    public OrderState apply(OrderEvent event) {
        switch (event.getType()) {
            case "OrderPlaced":
                this.status = "placed";
                this.total = event.getTotal();
                this.items = event.getItemIds();
                break;
            case "OrderPaid":
                this.status = "paid";
                break;
            case "OrderShipped":
                this.status = "shipped";
                break;
            case "OrderCancelled":
                this.status = "cancelled";
                break;
        }
        return this;
    }
}

Schema Registry con Avro (Python)

from confluent_kafka import Producer, SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from dataclasses import dataclass, asdict
import uuid

schema_registry_client = SchemaRegistryClient({
    'url': 'http://schema-registry:8081'
})

order_event_schema_str = """
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.stackpractices.events",
  "fields": [
    {"name": "event_id", "type": "string"},
    {"name": "event_type", "type": "string"},
    {"name": "aggregate_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "total", "type": "double"},
    {"name": "items", "type": {"type": "array", "items": "string"}},
    {"name": "occurred_at", "type": "string"}
  ]
}
"""

avro_serializer = AvroSerializer(
    schema_registry_client,
    order_event_schema_str,
    lambda obj, ctx: asdict(obj)
)

@dataclass
class OrderEvent:
    event_id: str
    event_type: str
    aggregate_id: str
    customer_id: str
    total: float
    items: list
    occurred_at: str

producer = SerializingProducer({
    'bootstrap.servers': 'kafka:9092',
    'value.serializer': avro_serializer,
})

def delivery_report(err, msg):
    # Se invoca cuando el broker confirma (o rechaza) el mensaje
    if err is not None:
        print(f"Entrega fallida: {err}")

def publish_order_event(order):
    event = OrderEvent(
        event_id=str(uuid.uuid4()),
        event_type='OrderPlaced',
        aggregate_id=order.id,
        customer_id=order.customer_id,
        total=order.total(),
        items=[item.id for item in order.items],
        occurred_at=order.created_at.isoformat(),
    )
    producer.produce(
        topic='orders',
        key=order.id.encode(),
        value=event,
        on_delivery=delivery_report,
    )
    producer.flush()

AWS EventBridge Consumer (TypeScript)

import { EventBridgeClient, PutEventsCommand } from '@aws-sdk/client-eventbridge';
import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs';

const eventBridge = new EventBridgeClient({ region: 'us-east-1' });
const sqs = new SQSClient({ region: 'us-east-1' });

// Publicar en EventBridge
async function publishOrderEvent(order: Order): Promise<void> {
  const command = new PutEventsCommand({
    Entries: [{
      EventBusName: 'stackpractices-events',
      Source: 'order-service',
      DetailType: 'OrderPlaced',
      Detail: JSON.stringify({
        orderId: order.id,
        customerId: order.customerId,
        total: order.total,
        items: order.items,
      }),
    }],
  });
  await eventBridge.send(command);
}

// Consumir desde SQS (target de EventBridge)
async function processOrderEvents(): Promise<void> {
  const result = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: process.env.INVENTORY_QUEUE_URL!,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,
  }));

  for (const message of result.Messages || []) {
    try {
      const event = JSON.parse(message.Body!);
      const detail = JSON.parse(event.detail);

      await reserveInventory(detail.items);
      await sqs.send(new DeleteMessageCommand({
        QueueUrl: process.env.INVENTORY_QUEUE_URL!,
        ReceiptHandle: message.ReceiptHandle!,
      }));
    } catch (error) {
      // El mensaje vuelve a ser visible tras el visibility timeout
      // Tras los reintentos máximos, pasa a la DLQ
      console.error('Fallo al procesar el evento:', error);
    }
  }
}

Explicación

• Evento vs command: un evento afirma que algo ocurrió (OrderPlaced); es inmutable y llega a todo el que escucha. Un command pide una acción (PlaceOrder) y va dirigido a un handler concreto por su nombre. No los mezcles — un servicio que recibe un command no debería republicarlo como evento sin transformarlo antes.
• Patrones de message broker: pub/sub reparte un mismo evento a todos los suscriptores, mientras que point-to-point entrega a un solo consumidor. Point-to-point escala metiendo más workers sobre la misma cola (competing consumers). Elige pub/sub si todos los servicios necesitan el evento; point-to-point si solo uno.
• Ordenamiento de eventos: los brokers no garantizan orden global de principio a fin. Si OrderPlaced y OrderCancelled llegan fuera de secuencia, el sistema de inventario puede intentar cancelar stock que nunca se reservó. Mantén el orden dentro del aggregate — enruta todos los eventos de order-123 a la misma partición — o haz los handlers idempotentes.
• Dead-letter queues: un mensaje que falla una y otra vez no debe quedarse bloqueando todo lo que viene detrás. Tras N reintentos, derívalo a una dead-letter queue para inspección manual. Vigila la profundidad de la DLQ como alerta crítica — una DLQ que crece suele indicar un problema sistémico, no mala suerte.

Variantes

Broker	Patrón	Durabilidad	Ordenamiento	Escala	Mejor para
Kafka	Pub/sub, streams	Alta	Por partición	Masiva	Event sourcing, streaming
RabbitMQ	Pub/sub, colas	Media	Por cola	Media	Enrutamiento complejo, AMQP
NATS	Pub/sub, request/reply	Baja	Ninguno	Muy alta	Baja latencia, simple
AWS SNS/SQS	Pub/sub, colas	Alta	Ninguno	Alta	Cloud-native, serverless
Redis Streams	Pub/sub	Media	Por stream	Media	Simple, Redis existente

Lo que funciona

• Diseña eventos, no mensajes: un evento describe qué ocurrió, no qué debe hacer el consumidor. Consulta Microservices Patterns para estrategias de comunicación entre servicios. OrderPlaced es correcto. DecrementInventory es un command disfrazado de evento. Los eventos son hechos; los commands son instrucciones.
• Usa validación de esquemas: los eventos sin validar son fuente de bugs sutiles. Usa Avro, JSON Schema o Protobuf para definir contratos de eventos. Valida en los límites de publisher y consumer. Versiona los esquemas y mantén retrocompatibilidad — un Schema Registry lo hace de forma centralizada.
• Haz consumidores idempotentes: entre los reintentos de red y las reentregas del broker, el mismo evento acaba llegando más de una vez. Consulta Endpoints Idempotentes para patrones de deduplicación. Diseña handlers para que procesar el mismo evento dos veces produzca el mismo estado. Usa UPSERT o registra los IDs de evento procesados en una tabla de deduplicación.
• Monitoriza el consumer lag: el lag es el número de mensajes sin procesar en una partición. Si el lag no para de subir, tus consumidores no dan abasto con los productores. Alerta sobre umbrales de lag y después escala consumidores u optimiza el handler.
• Publica eventos de dominio, no de infraestructura: PaymentProcessed es un evento de dominio con significado de negocio. DatabaseRowInserted es ruido de infraestructura. A los consumidores les importan los cambios de estado del negocio, no los detalles de implementación.

Errores comunes

• Coreografía sin visibilidad: un request que se despliega en 5 eventos, cada uno disparando 3 más, crea un workflow invisible. Cuando falla, depurar significa revisar 15 servicios a mano. Añade correlation IDs y distributed tracing para seguir la cadena.
• Procesamiento síncrono de eventos: un consumer que procesa eventos de forma síncrona dentro de un HTTP request reintroduce el acoplamiento que el event bus debía eliminar. Procesa los eventos asíncronamente, desacoplados del request del usuario.
• Sin manejo de error para mensajes envenenados: un evento malformado que tumba al consumer vuelve a la cola una y otra vez y acaba bloqueándola. Define un máximo de reintentos y deriva los fallos a un manejador de poison pills.
• Guardar estado en el broker: usar el broker como base de datos (consultar Kafka por el estado actual) es un anti-pattern. Los brokers son para transporte, no para almacenamiento. Usa event sourcing o un read model para las consultas de estado.

Ver También

• Saga Pattern — transacciones compensatorias cuando un workflow abarca varios servicios.
• CQRS Pattern — separa los modelos de lectura y escritura cuando los eventos se convierten en la fuente de verdad.
• Especificación CloudEvents — un formato de envoltura común si quieres eventos portables entre brokers.
• Código complementario en GitHub — el repositorio companion de esta receta.