Patrón Publish-Subscribe

Descripción General

Una cola punto a punto entrega cada mensaje a un único consumidor. Eso funciona para repartir trabajo, pero muchos sistemas necesitan varias reacciones independientes al mismo evento. Cuando se crea un pedido, el servicio de inventario actualiza el stock, el de notificaciones envía el email de confirmación y el de analítica registra el evento: tres consumidores, un mensaje.

El patrón Publish-Subscribe resuelve esto difundiendo cada mensaje a todos los suscriptores de un tópico. Cada suscriptor recibe su propia copia y la procesa a su ritmo, mientras el publicador ignora quién está escuchando.

Cuándo Usar

Pub/sub encaja cuando:

• Varios servicios necesitan reaccionar al mismo evento de forma independiente.
• Quieres desacoplar el productor de los consumidores: añadir suscriptores no debería requerir cambios aguas arriba.
• Los consumidores procesan el evento a velocidades distintas o con lógica diferente.
• El evento importa a más de un contexto delimitado (pedidos → inventario + facturación + analítica).

No encaja cuando:

• Un solo consumidor debe manejar cada mensaje; eso es una cola de trabajo, y pub/sub duplicaría el esfuerzo.
• Necesitas semántica de petición/respuesta o un valor de retorno; pub/sub es fire-and-forget.
• Todos los suscriptores deben ver los mensajes en un orden global estricto, ya que la mayoría de brokers solo garantiza orden por partición o por canal.
• El productor necesita saber si alguien consumió el evento; no hay acuse de recibo hacia el publicador.

Para mantener suscriptores sincronizados mediante datos en lugar de notificaciones, el patrón Event-Carried State Transfer encaja mejor. Para la arquitectura completa alrededor de los flujos de eventos, consulta la guía de sistemas orientados a eventos.

Solución

La forma siempre es la misma: un publicador escribe en un tópico y el broker difunde el mensaje a cada suscriptor:

Python (Pub/Sub de Redis)

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"Publicado: {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"[Inventario] Actualizando stock del pedido {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"[Notificaciones] Enviando confirmación del pedido {event['order_id']}")

# Arrancar suscriptores en hilos
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)

Redis Pub/Sub es la opción más simple pero es fire-and-forget: un suscriptor desconectado cuando publicas pierde el mensaje. Para una versión con garantías de entrega en producción, la receta de pub/sub con Redis y la receta de streaming de eventos con Kafka cubren las configuraciones durables.

JavaScript (RabbitMQ exchange fanout)

import amqp from "amqplib";

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

  // El exchange fanout 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 las colas al exchange fanout
  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(`Publicado: pedido ${orderId}`);
  }

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

  // Suscriptor 2: Notificaciones
  channel.consume(notificationQueue, (msg) => {
    if (!msg) return;
    const event = JSON.parse(msg.content.toString());
    console.log(`[Notificaciones] Enviando confirmación del pedido ${event.orderId}`);
    channel.ack(msg);
  });

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

setup();

Java (Spring + tópicos de Kafka)

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("Publicado: pedido " + orderId);
        } catch (Exception e) {
            throw new RuntimeException("Error al publicar", 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("[Inventario] Actualizando stock del pedido " + 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("[Notificaciones] Enviando confirmación del pedido " + event.getOrderId());
        }
    }

    private OrderEvent parseEvent(String message) {
        try {
            return objectMapper.readValue(message, OrderEvent.class);
        } catch (Exception e) {
            throw new RuntimeException("Error al parsear el evento", e);
        }
    }
}

Event Bus en Memoria (TypeScript)

Dentro de un solo proceso no hace falta un broker. Un pequeño bus de eventos te da el mismo desacoplamiento:

// 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);
    // Devuelve la función para desuscribirse
    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] Error en handler de ${event}:`, err); }
    });
  }
}

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

// Uso: múltiples suscriptores desacoplados
const bus = new EventBus();

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

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

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

// Suscriptor 4: notificar envío
bus.subscribe("order.shipped", (data: OrderShipped) => {
  smsService.send(data.orderId, `Enviado: ${data.trackingNumber}`);
});

// Publicador: 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 });
}

Mismo patrón, distinta escala: un bus en memoria para un proceso, Redis para mensajería barata entre servicios, RabbitMQ o Kafka cuando importa la durabilidad.

Explicación

El publicador envía mensajes a un tópico (o exchange) sin saber qué suscriptores existen. El broker mantiene una lista de suscripciones por tópico y entrega una copia a cada uno. Cómo llega la copia depende del broker:

• Exchange fanout (RabbitMQ): empuja cada mensaje a todas las colas enlazadas al exchange. Cada cola guarda su propia copia, así que cada consumidor lee de forma independiente.
• Consumer groups de Kafka: cada grupo recibe su propia copia de cada mensaje. Dentro del grupo, las particiones se reparten entre consumidores para paralelizar, así que el “suscriptor” es el grupo, no un consumidor individual.
• Pub/Sub de Redis: entrega fire-and-forget a los listeners conectados. Sin almacenamiento, sin replay, sin durabilidad.

La ventaja es el desacoplamiento: al publicador le da igual quién se suscribe. Añadir un consumidor de analítica es crear una cola y enlazarla, con cero cambios aguas arriba. El coste es la observabilidad: sin vínculo directo entre productor y consumidores, hay que vigilar suscripciones y profundidad de cola para notar un suscriptor muerto o retrasado.

Variantes

Variante	Mecanismo	Caso de Uso	Compromiso
Exchange Fanout	Fanout de RabbitMQ	Todos los suscriptores reciben todo	Sin filtrado a nivel de broker
Exchange Topic	Topic de RabbitMQ con routing keys	Suscriptores filtran por patrón	Enrutado más complejo
Consumer Groups de Kafka	Tópicos de Kafka + grupos	Durable, particionado, reproducible	Más infraestructura
SNS + SQS	Fanout de AWS SNS a colas SQS	Gestionado, durable por suscriptor	Solo AWS, coste por mensaje
Pub/Sub de Redis	Canales de Redis	Simple, baja latencia	Sin durabilidad, mensajes perdidos si está offline

Pub/Sub vs Observer

Los dos se confunden constantemente. Observer es directo: el sujeto guarda referencias a sus observadores y los llama: mismo proceso, mismo grafo de objetos. Pub/Sub pone un broker (o bus) en medio, así que publicador y suscriptores no saben que el otro existe.

Usa Observer para notificaciones dentro de un módulo (un formulario que reacciona a cambios de input, un widget de UI que observa un modelo). Usa Pub/Sub cuando cruzas fronteras de servicio o proceso, o cuando quieres añadir consumidores sin tocar el publicador. Observer es más simple; Pub/Sub escala más lejos.

Qué Funciona

• Usa colas durables para cada suscriptor para que los mensajes sobrevivan reinicios del broker.
• Dale a cada suscriptor su propia cola (nunca compartida) para que procesen de forma independiente.
• Usa consumer groups en Kafka para escalar dentro de un servicio suscriptor.
• Incluye el tipo de evento en el mensaje para que los suscriptores filtren lo que les interesa.
• Haz los suscriptores idempotentes; el mismo evento puede entregarse más de una vez.
• Vigila el retraso de los suscriptores; un consumidor lento acumula una cola enorme en silencio.
• Dale a cada suscriptor su propia dead-letter queue para que uno que falla no envenene al resto (el runbook de dead-letter queues tiene el checklist operativo).

Errores Comunes

• Cola compartida entre varios suscriptores: los suscriptores compiten por los mensajes en lugar de recibir cada uno su copia. Cada suscriptor necesita su propia cola.
• Redis Pub/Sub para eventos críticos: los mensajes se pierden cuando un suscriptor está offline. Usa Redis Streams o un broker durable para eventos que no pueden perderse.
• Suscriptores síncronos que bloquean al publicador: si el publicador espera el acuse de los suscriptores, los lentos lo frenan. Mantén la entrega asíncrona.
• Sin filtrado por tipo de evento: los suscriptores procesan todos los eventos aunque solo les interese uno. Usa exchanges topic o filtra en el consumidor.
• Añadir suscriptores sin escalar el broker: más suscriptores significa más copias de cada mensaje. Comprueba que el broker aguanta el throughput extra.
• Ignorar consumidores lentos: un suscriptor que no da abasto acumula mensajes. Define una profundidad máxima de cola y alerta sobre ella.

Ver También

• Guía de Sistemas Orientados a Eventos
• Patrón Event-Carried State Transfer
• Receta de Pub/Sub con Redis
• Runbook de Dead Letter Queues