Patrón Publish-Subscribe

Descripción General

En una cola punto-a-punto, cada mensaje es procesado por exactamente un consumidor. Pero muchos sistemas necesitan multiples reacciones independientes al mismo evento. Cuando se realiza un pedido, el servicio de inventario actualiza stock, el servicio de notificaciones envia email de confirmacion, y el servicio de analitica registra el evento. El patron Publish-Subscribe difunde un mensaje a todos los suscriptores de un topico, para que cada suscriptor reciba su propia copia y la procese independientemente.

Cuándo Usar

• For alternatives, see Event-Carried State Transfer Pattern.

• Multiples servicios necesitan reaccionar al mismo evento independientemente

• Quieres desacoplar al productor de los consumidores (el productor no sabe quien consume)

• Nuevos consumidores pueden agregarse sin modificar al productor

• Diferentes consumidores procesan el evento a diferentes velocidades o con diferente logica

Solución

Python (Redis Pub/Sub)

import redis
import json
import threading

r = redis.Redis(host="localhost", port=6379)

# Publicador
def publish_order(order_id, total):
    event = {"type": "order.created", "order_id": order_id, "total": total}
    r.publish("order.events", json.dumps(event))
    print(f"Published: {event}")

# Suscriptor 1: Servicio de inventario
def inventory_subscriber():
    pubsub = r.pubsub()
    pubsub.subscribe("order.events")
    for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            if event["type"] == "order.created":
                print(f"[Inventory] Updating stock for order {event['order_id']}")

# Suscriptor 2: Servicio de notificaciones
def notification_subscriber():
    pubsub = r.pubsub()
    pubsub.subscribe("order.events")
    for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            if event["type"] == "order.created":
                print(f"[Notification] Sending confirmation for order {event['order_id']}")

# Iniciar suscriptores en threads
threading.Thread(target=inventory_subscriber, daemon=True).start()
threading.Thread(target=notification_subscriber, daemon=True).start()

# Publicar eventos
publish_order(1, 99.99)
publish_order(2, 49.99)

JavaScript (RabbitMQ — fanout exchange)

import amqp from "amqplib";

async function setup() {
  const conn = await amqp.connect("amqp://localhost");
  const channel = await conn.createChannel();

  // Fanout exchange difunde a todas las colas enlazadas
  await channel.assertExchange("order.events", "fanout", { durable: true });

  // Crear colas para cada suscriptor
  const { queue: inventoryQueue } = await channel.assertQueue("inventory", {
    durable: true,
  });
  const { queue: notificationQueue } = await channel.assertQueue("notification", {
    durable: true,
  });

  // Enlazar colas al fanout exchange
  await channel.bindQueue(inventoryQueue, "order.events", "");
  await channel.bindQueue(notificationQueue, "order.events", "");

  // Publicador
  async function publishOrder(orderId, total) {
    const event = JSON.stringify({ type: "order.created", orderId, total });
    channel.publish("order.events", "", Buffer.from(event), {
      persistent: true,
    });
    console.log(`Published: order ${orderId}`);
  }

  // Suscriptor 1: Inventario
  channel.consume(inventoryQueue, (msg) => {
    if (!msg) return;
    const event = JSON.parse(msg.content.toString());
    console.log(`[Inventory] Updating stock for order ${event.orderId}`);
    channel.ack(msg);
  });

  // Suscriptor 2: Notificaciones
  channel.consume(notificationQueue, (msg) => {
    if (!msg) return;
    const event = JSON.parse(msg.content.toString());
    console.log(`[Notification] Sending confirmation for order ${event.orderId}`);
    channel.ack(msg);
  });

  // Publicar eventos
  await publishOrder(1, 99.99);
  await publishOrder(2, 49.99);
}

setup();

Java (Spring + Kafka topics)

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;

@Component
public class OrderEventPublisher {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper = new ObjectMapper();

    public OrderEventPublisher(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Publicador
    public void publishOrder(Long orderId, double total) {
        try {
            String event = objectMapper.writeValueAsString(
                new OrderEvent("order.created", orderId, total)
            );
            kafkaTemplate.send("order-events", String.valueOf(orderId), event);
            System.out.println("Published: order " + orderId);
        } catch (Exception e) {
            throw new RuntimeException("Failed to publish", e);
        }
    }

    // Suscriptor 1: Inventario
    @KafkaListener(topics = "order-events", groupId = "inventory-service")
    public void handleInventory(String message) {
        OrderEvent event = parseEvent(message);
        if ("order.created".equals(event.getType())) {
            System.out.println("[Inventory] Updating stock for order " + event.getOrderId());
        }
    }

    // Suscriptor 2: Notificaciones
    @KafkaListener(topics = "order-events", groupId = "notification-service")
    public void handleNotification(String message) {
        OrderEvent event = parseEvent(message);
        if ("order.created".equals(event.getType())) {
            System.out.println("[Notification] Sending confirmation for order " + event.getOrderId());
        }
    }

    private OrderEvent parseEvent(String message) {
        try {
            return objectMapper.readValue(message, OrderEvent.class);
        } catch (Exception e) {
            throw new RuntimeException("Failed to parse event", e);
        }
    }
}

Explicación

El publicador envia mensajes a un topico (o exchange) sin saber que suscriptores existen. El broker mantiene una lista de suscripciones para cada topico. Cuando llega un mensaje, el broker entrega una copia a cada suscriptor.

Fanout exchange (RabbitMQ): Difunde cada mensaje a todas las colas enlazadas al exchange. Cada cola obtiene su propia copia, y cada consumidor lee de su cola independientemente.

Kafka consumer groups: Kafka usa consumer groups para implementar pub/sub. Cada consumer group obtiene su propia copia de cada mensaje. Dentro de un grupo, las particiones se distribuyen entre consumidores para paralelismo.

Redis Pub/Sub: Fire-and-forget. Si un suscriptor no esta escuchando cuando se publica un mensaje, lo pierde. Usa Redis Streams en su lugar para durabilidad.

El beneficio clave es acoplamiento ligero: el publicador no sabe ni le importan los suscriptores. Agregar un nuevo suscriptor (ej. un servicio de analitica) no requiere cambios al publicador ni a otros suscriptores.

Variantes

Variante	Mecanismo	Caso de Uso	Compromiso
Fanout Exchange	RabbitMQ fanout	Todos los suscriptores reciben todos los mensajes	Sin filtrado a nivel broker
Topic Exchange	RabbitMQ topic con routing keys	Suscriptores filtran por patron	Ruteo mas complejo
Kafka Consumer Groups	Kafka topics + groups	Durable, particionado, reejecutable	Mas infraestructura
SNS + SQS	AWS SNS fanout a SQS queues	Administrado, durable por suscriptor	Solo AWS, costo por mensaje
Redis Pub/Sub	Redis channels	Simple, baja latencia	Sin durabilidad, mensajes perdidos si offline

Qué Funciona

• Usa colas durables para cada suscriptor para que los mensajes sobrevivan reinicios del broker
• Dale a cada suscriptor su propia cola (no compartida) para que procesen independientemente
• Usa consumer groups en Kafka para escalar dentro de un servicio suscriptor
• Incluye tipo de evento en el mensaje para que los suscriptores puedan filtrar eventos relevantes
• Haz suscriptores idempotentes — el mismo evento puede entregarse mas de una vez
• Monitorea el lag del suscriptor — un suscriptor lento puede acumular una cola grande
• Usa dead-letter queues por suscriptor para que un suscriptor fallido no afecte a otros

Errores Comunes

• Cola compartida para multiples suscriptores: Todos los suscriptores compiten por mensajes en lugar de cada uno obtener una copia. Cada suscriptor necesita su propia cola.
• Redis Pub/Sub para eventos criticos: Los mensajes se pierden si un suscriptor esta offline.
• Suscriptores sincronos bloqueando al publicador: Si el publicador espera acknowledgments de suscriptores, los suscriptores lentos bloquean al publicador.
• Sin filtrado por tipo de evento: Los suscriptores procesan cada evento incluso si solo les importa un tipo.
• Agregar suscriptores sin escalar el broker: Mas suscriptores significa mas copias de mensajes.
• No manejar consumidores lentos: Un suscriptor que no puede mantener el ritmo acumula mensajes. Establece max depth de cola y alertas.

Temas Avanzados

Escenario: Pub/Sub para Notificaciones de Pedidos

// Pub/Sub: desacoplar productor de consumidores
class EventBus {
  private subscribers = new Map<string, Set<(data: unknown) => void>>();

  subscribe(event: string, handler: (data: unknown) => void): () => void {
    if (!this.subscribers.has(event)) this.subscribers.set(event, new Set());
    this.subscribers.get(event)!.add(handler);
    // Retornar funcion de unsubscribe
    return () => this.subscribers.get(event)?.delete(handler);
  }

  publish(event: string, data: unknown): void {
    this.subscribers.get(event)?.forEach(handler => {
      try { handler(data); }
      catch (err) { console.error(`[PUB/SUB] Handler error for ${event}:`, err); }
    });
  }
}

// Eventos del dominio
interface OrderCreated { orderId: string; customerEmail: string; total: number; }
interface OrderShipped { orderId: string; trackingNumber: string; }
interface OrderCancelled { orderId: string; reason: string; }

// Uso: multiples subscribers desacoplados
const bus = new EventBus();

// Subscriber 1: enviar email
bus.subscribe("order.created", (data: OrderCreated) => {
  emailService.send(data.customerEmail, "Order confirmed", `Total: ${data.total}`);
});

// Subscriber 2: actualizar analytics
bus.subscribe("order.created", (data: OrderCreated) => {
  analytics.track("order_created", { total: data.total });
});

// Subscriber 3: invalidar cache
bus.subscribe("order.created", () => {
  cache.invalidate("orders:recent");
});

// Subscriber 4: notificar shipping
bus.subscribe("order.shipped", (data: OrderShipped) => {
  smsService.send(data.orderId, `Shipped: ${data.trackingNumber}`);
});

// Publisher: el servicio de pedidos
async function createOrder(order: Order) {
  await db.save(order);
  bus.publish("order.created", { orderId: order.id, customerEmail: order.email, total: order.total });
}

// El publisher no conoce los subscribers

Lecciones:

• Pub/Sub desacopla productor de consumidores
• El publisher no conoce quienes son los subscribers
• Multiples subscribers reaccionan al mismo evento
• Anadir nuevo subscriber no requiere cambiar el publisher
• En sistemas distribuidos: usar Kafka, RabbitMQ, SNS+SQS
• En memoria: EventEmitter o EventBus custom

### Pub/Sub vs Observer: cual uso?

Pub/Sub es mas amplio: el publisher no conoce los subscribers, hay un bus/broker en medio. Observer es mas directo: el sujeto conoce a los observadores. Usa Pub/Sub para sistemas distribuidos o cuando quieres desacoplamiento total. Usa Observer para notificaciones dentro de un mismo modulo. Pub/Sub escala mejor: el broker enruta eventos. Observer es mas simple: no hay broker. Para microservicios, Pub/Sub. Para reactive forms, Observer.