Publish-Subscribe Pattern

Overview

A point-to-point queue delivers each message to exactly one consumer. That’s fine for distributing work, but many systems need several independent reactions to the same event. When an order is placed, the inventory service updates stock, the notification service sends a confirmation email, and the analytics service records the event: three consumers, one message.

The Publish-Subscribe pattern solves this by broadcasting each message to every subscriber of a topic. Each subscriber gets its own copy and processes it at its own pace, and the publisher stays unaware of who’s listening.

When to Use

Pub/sub fits when:

• Multiple services need to react to the same event independently.
• You want the producer decoupled from the consumers: new subscribers shouldn’t require changes upstream.
• Consumers process the event at different speeds or with different logic.
• The event matters to more than one bounded context (orders → inventory + billing + analytics).

It does not fit when:

• A single consumer should handle each message; that’s a work queue, and pub/sub would duplicate the work.
• You need request/response semantics or a return value; pub/sub is fire-and-forget.
• Every subscriber must see messages in strict global order, since most brokers only guarantee per-partition or per-channel ordering.
• The producer needs to know whether anyone consumed the event; there’s no built-in acknowledgment back to the publisher.

For keeping subscribers synced via data payloads rather than notifications, the Event-Carried State Transfer pattern is a better fit. For the broader architecture around event flows, see the event-driven systems guide.

Solution

The shape is always the same: a publisher writes to a topic, and the broker fans the message out to every subscriber:

Python (Redis Pub/Sub)

import redis
import json
import threading

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

# Publisher
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}")

# Subscriber 1: Inventory service
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']}")

# Subscriber 2: Notification service
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']}")

# Start subscribers in threads
threading.Thread(target=inventory_subscriber, daemon=True).start()
threading.Thread(target=notification_subscriber, daemon=True).start()

# Publish events
publish_order(1, 99.99)
publish_order(2, 49.99)

Redis Pub/Sub is the simplest option but it’s fire-and-forget: a subscriber that isn’t connected when you publish loses the message. For a production-grade version with delivery guarantees, the Redis pub/sub recipe and the Kafka event streaming recipe cover the durable setups.

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 broadcasts to all bound queues
  await channel.assertExchange("order.events", "fanout", { durable: true });

  // Create queues for each subscriber
  const { queue: inventoryQueue } = await channel.assertQueue("inventory", {
    durable: true,
  });
  const { queue: notificationQueue } = await channel.assertQueue("notification", {
    durable: true,
  });

  // Bind queues to the fanout exchange
  await channel.bindQueue(inventoryQueue, "order.events", "");
  await channel.bindQueue(notificationQueue, "order.events", "");

  // Publisher
  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}`);
  }

  // Subscriber 1: Inventory
  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);
  });

  // Subscriber 2: Notification
  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);
  });

  // Publish events
  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;
    }

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

    // Subscriber 1: Inventory
    @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());
        }
    }

    // Subscriber 2: Notification
    @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);
        }
    }
}

In-Memory Event Bus (TypeScript)

Inside a single process you don’t need a broker at all. A small event bus gives you the same decoupling:

// Pub/Sub: decouple producer from consumers
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);
    // Return unsubscribe function
    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); }
    });
  }
}

// Domain events
interface OrderCreated { orderId: string; customerEmail: string; total: number; }
interface OrderShipped { orderId: string; trackingNumber: string; }
interface OrderCancelled { orderId: string; reason: string; }

// Usage: multiple decoupled subscribers
const bus = new EventBus();

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

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

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

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

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

Same pattern, different scale: an in-memory bus for one process, Redis for cheap cross-service messaging, RabbitMQ or Kafka when durability matters.

Explanation

The publisher sends messages to a topic (or exchange) without knowing which subscribers exist. The broker keeps a subscription list per topic and delivers a copy to each one. How the copy gets there depends on the broker:

• Fanout exchange (RabbitMQ): pushes every message to all queues bound to the exchange. Each queue holds its own copy, so each consumer reads independently.
• Kafka consumer groups: each group gets its own copy of every message. Within a group, partitions are split across consumers for parallelism, so a group is the “subscriber,” not an individual consumer.
• Redis Pub/Sub: fire-and-forget delivery to connected listeners. No storage, no replay, no durability.

The payoff is loose coupling: the publisher doesn’t know or care about subscribers. Adding an analytics consumer means creating a queue and binding it, with zero changes upstream. The cost is observability: with no direct link between producer and consumers, you have to monitor subscriptions and queue depth to notice a dead or lagging subscriber.

Variants

Variant	Mechanism	Use Case	Tradeoff
Fanout Exchange	RabbitMQ fanout	All subscribers get all messages	No filtering at broker level
Topic Exchange	RabbitMQ topic with routing keys	Subscribers filter by pattern	More complex routing
Kafka Consumer Groups	Kafka topics + groups	Durable, partitioned, replayable	More infrastructure
SNS + SQS	AWS SNS fanout to SQS queues	Managed, durable per-subscriber	AWS-only, cost per message
Redis Pub/Sub	Redis channels	Simple, low latency	No durability, messages lost if offline

Pub/Sub vs Observer

The two get confused constantly. Observer is direct: the subject keeps a reference to its observers and calls them: same process, same object graph. Pub/Sub puts a broker (or bus) in the middle, so publisher and subscribers don’t know each other exists.

Use Observer for notifications inside one module (a form reacting to input changes, a UI widget watching a model). Use Pub/Sub when you’re crossing service or process boundaries, or when you want to add consumers without touching the publisher. Observer is simpler; Pub/Sub scales further.

What Works

• Use durable queues for each subscriber so messages survive broker restarts.
• Give each subscriber its own queue (never shared) so they process independently.
• Use consumer groups in Kafka to scale within a subscriber service.
• Include the event type in the message so subscribers can filter what they care about.
• Make subscribers idempotent; the same event can be delivered more than once.
• Watch subscriber lag; a slow consumer quietly builds up a large queue.
• Give each subscriber its own dead-letter queue so a failing one doesn’t poison the rest (the dead-letter queue runbook has the operational checklist).

Common Mistakes

• Shared queue for multiple subscribers: subscribers compete for messages instead of each getting a copy. Every subscriber needs its own queue.
• Redis Pub/Sub for critical events: messages are lost when a subscriber is offline. Use Redis Streams or a durable broker for events that must not be lost.
• Synchronous subscribers blocking the publisher: if the publisher waits for subscriber acknowledgments, slow subscribers stall it. Keep delivery async.
• No filtering by event type: subscribers chew through every event even when they only care about one. Use topic exchanges or filter in the consumer.
• Adding subscribers without scaling the broker: more subscribers means more message copies. Check the broker can absorb the extra throughput.
• Ignoring slow consumers: a subscriber that can’t keep up accumulates messages. Set a max queue depth and alert on it.

See Also

• Event-Driven Systems Guide
• Event-Carried State Transfer Pattern
• Redis Pub/Sub Recipe
• Dead Letter Queue Runbook