Design Patterns, Explained Simply · Modern Patterns and Choosing Well
Pub/Sub and the Event Bus
Decouple publishers from subscribers completely with named topics and a broker in between, so services can react to events without knowing who produced them.
The problem
The Observer lesson showed an order subject calling its subscribers directly. That works inside one process, but:
- the order code still holds references to every listener,
- a slow listener (sending email) makes order placement slow,
- if a listener throws, the order request fails,
- the shipping service runs in a different process, or a different language, and can't register a callback.
Real-world analogy
A radio station broadcasts on a frequency (the topic). It doesn't know who is listening; listeners tune in or out whenever they like and never contact the station. If nobody is listening, the broadcast still goes out. With a recording device (a durable queue), you can even listen later.
Without the pattern
class OrderService:
def __init__(self, email, inventory, warehouse, analytics): # knows every consumer
...
def place(self, order):
self.email.send(order) # slow: the customer waits for the email
self.warehouse.notify(order) # if this throws, the order request fails
The pattern
- Publisher: sends a message to a topic (
"order.placed"), not to a subscriber. - Subscribers: register interest in topics with the broker or event bus.
- Broker / event bus: routes each message to the subscribers of its topic. It is often asynchronous (a queue), so publishers don't wait, and failures are isolated (retries, dead-letter queues).
Observer vs Pub/Sub:
| Observer | Pub/Sub | |
|---|---|---|
| Who knows whom | Subject holds its observers | Neither side knows the other; both know the broker and topic |
| Delivery | Usually synchronous, same process | Usually asynchronous, can cross processes and services |
| Failure | One observer's error can affect the subject | Isolated; retried or dead-lettered by the broker |
The code
An in-memory event bus with topics, asynchronous delivery (messages are queued and delivered by drain()), and a dead-letter list for a failing subscriber.
class EventBus:
def __init__(self):
self.subscribers, self.queue, self.dead_letters = {}, [], []
def subscribe(self, topic, name, handler):
self.subscribers.setdefault(topic, []).append((name, handler))
def publish(self, topic, payload): # returns immediately: just queues the message
self.queue.append((topic, payload))
def drain(self): # in real systems a broker and workers do this
while self.queue:
topic, payload = self.queue.pop(0)
for name, handler in self.subscribers.get(topic, []):
try:
handler(payload)
except Exception as e:
self.dead_letters.append(f"{name} failed on {topic}: {e}")
bus = EventBus()
def email_service(order):
print(f"email: receipt for order {order['id']}")
def warehouse_service(order):
if order["id"] == 2:
raise RuntimeError("warehouse API down")
print(f"warehouse: pick order {order['id']}")
bus.subscribe("order.placed", "email", email_service)
bus.subscribe("order.placed", "warehouse", warehouse_service)
bus.subscribe("user.signed_up", "email", lambda u: print(f"email: welcome {u['name']}"))
bus.publish("order.placed", {"id": 1}) # the order service only knows the bus
bus.publish("order.placed", {"id": 2})
bus.publish("user.signed_up", {"name": "Ada"})
print("published 3 events, delivered so far: 0")
bus.drain()
print("dead letters:", bus.dead_letters)
Output
published 3 events, delivered so far: 0 email: receipt for order 1 warehouse: pick order 1 email: receipt for order 2 email: welcome Ada dead letters: ['warehouse failed on order.placed: warehouse API down']
type Handler = (payload: Record<string, unknown>) => void;
class EventBus {
private subscribers = new Map<string, [string, Handler][]>();
private queue: [string, Record<string, unknown>][] = [];
deadLetters: string[] = [];
subscribe(topic: string, name: string, handler: Handler) {
if (!this.subscribers.has(topic)) this.subscribers.set(topic, []);
this.subscribers.get(topic)!.push([name, handler]);
}
publish(topic: string, payload: Record<string, unknown>) { // returns immediately: just queues the message
this.queue.push([topic, payload]);
}
drain() { // in real systems a broker and workers do this
while (this.queue.length) {
const [topic, payload] = this.queue.shift()!;
for (const [name, handler] of this.subscribers.get(topic) ?? []) {
try {
handler(payload);
} catch (e) {
this.deadLetters.push(`${name} failed on ${topic}: ${(e as Error).message}`);
}
}
}
}
}
const bus = new EventBus();
bus.subscribe("order.placed", "email", (o) => console.log(`email: receipt for order ${o.id}`));
bus.subscribe("order.placed", "warehouse", (o) => {
if (o.id === 2) throw new Error("warehouse API down");
console.log(`warehouse: pick order ${o.id}`);
});
bus.subscribe("user.signed_up", "email", (u) => console.log(`email: welcome ${u.name}`));
bus.publish("order.placed", { id: 1 }); // the order service only knows the bus
bus.publish("order.placed", { id: 2 });
bus.publish("user.signed_up", { name: "Ada" });
console.log("published 3 events, delivered so far: 0");
bus.drain();
console.log(`dead letters: [${bus.deadLetters.map((d) => `'${d}'`).join(", ")}]`);
Output
published 3 events, delivered so far: 0 email: receipt for order 1 warehouse: pick order 1 email: receipt for order 2 email: welcome Ada dead letters: ['warehouse failed on order.placed: warehouse API down']
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;
public class Main {
public static void main(String[] args) {
EventBus bus = new EventBus();
bus.subscribe("order.placed", "email", o -> System.out.println("email: receipt for order " + o.get("id")));
bus.subscribe("order.placed", "warehouse", o -> {
if (o.get("id").equals("2")) throw new RuntimeException("warehouse API down");
System.out.println("warehouse: pick order " + o.get("id"));
});
bus.subscribe("user.signed_up", "email", u -> System.out.println("email: welcome " + u.get("name")));
bus.publish("order.placed", Map.of("id", "1")); // the order service only knows the bus
bus.publish("order.placed", Map.of("id", "2"));
bus.publish("user.signed_up", Map.of("name", "Ada"));
System.out.println("published 3 events, delivered so far: 0");
bus.drain();
System.out.println("dead letters: " + bus.deadLetters.stream().map(d -> "'" + d + "'")
.collect(Collectors.joining(", ", "[", "]")));
}
}
class EventBus {
private record Sub(String name, Consumer<Map<String, String>> handler) {}
private record Msg(String topic, Map<String, String> payload) {}
private final Map<String, List<Sub>> subscribers = new HashMap<>();
private final Deque<Msg> queue = new ArrayDeque<>();
final List<String> deadLetters = new ArrayList<>();
void subscribe(String topic, String name, Consumer<Map<String, String>> handler) {
subscribers.computeIfAbsent(topic, t -> new ArrayList<>()).add(new Sub(name, handler));
}
void publish(String topic, Map<String, String> payload) { queue.add(new Msg(topic, payload)); } // just queues
void drain() { // in real systems a broker and workers do this
while (!queue.isEmpty()) {
Msg msg = queue.poll();
for (Sub sub : subscribers.getOrDefault(msg.topic(), List.of())) {
try {
sub.handler().accept(msg.payload());
} catch (RuntimeException e) {
deadLetters.add(sub.name() + " failed on " + msg.topic() + ": " + e.getMessage());
}
}
}
}
}
Output
published 3 events, delivered so far: 0 email: receipt for order 1 warehouse: pick order 1 email: receipt for order 2 email: welcome Ada dead letters: ['warehouse failed on order.placed: warehouse API down']
Run this on your computer — see “Get Started” for setup.
<?php
class EventBus {
private array $subscribers = [];
private array $queue = [];
public array $deadLetters = [];
public function subscribe(string $topic, string $name, callable $handler): void {
$this->subscribers[$topic][] = [$name, $handler];
}
public function publish(string $topic, array $payload): void { // returns immediately: just queues the message
$this->queue[] = [$topic, $payload];
}
public function drain(): void { // in real systems a broker and workers do this
while ($this->queue) {
[$topic, $payload] = array_shift($this->queue);
foreach ($this->subscribers[$topic] ?? [] as [$name, $handler]) {
try {
$handler($payload);
} catch (Throwable $e) {
$this->deadLetters[] = "$name failed on $topic: {$e->getMessage()}";
}
}
}
}
}
$bus = new EventBus();
$bus->subscribe("order.placed", "email", fn($o) => print("email: receipt for order {$o['id']}\n"));
$bus->subscribe("order.placed", "warehouse", function ($o) {
if ($o["id"] === 2) throw new RuntimeException("warehouse API down");
echo "warehouse: pick order {$o['id']}\n";
});
$bus->subscribe("user.signed_up", "email", fn($u) => print("email: welcome {$u['name']}\n"));
$bus->publish("order.placed", ["id" => 1]); // the order service only knows the bus
$bus->publish("order.placed", ["id" => 2]);
$bus->publish("user.signed_up", ["name" => "Ada"]);
echo "published 3 events, delivered so far: 0\n";
$bus->drain();
echo "dead letters: [", implode(", ", array_map(fn($d) => "'$d'", $bus->deadLetters)), "]\n";
Output
published 3 events, delivered so far: 0 email: receipt for order 1 warehouse: pick order 1 email: receipt for order 2 email: welcome Ada dead letters: ['warehouse failed on order.placed: warehouse API down']
Publishing returned immediately, the warehouse failure didn't affect the email service or the publisher, and the failed delivery was kept for a retry.
When to use it
- Services or modules should react to events without the source knowing them: microservices, modular monoliths, background processing.
- Reactions are slow or unreliable (emails, third-party APIs) and must not slow down or break the main action.
- Many independent consumers, possibly added later, care about the same events.
When not to use it
- You need the result immediately in the same request. Use a direct call.
- One simple in-process reaction. Observer or a direct call is easier to read and debug.
- Asynchronous messaging brings eventual consistency, duplicate deliveries (make handlers idempotent), ordering questions and harder debugging. Use it where those trade-offs are worth it.
You've already used it
- Brokers: Amazon SNS/SQS, Google Pub/Sub, Kafka, RabbitMQ, Redis Pub/Sub, NATS.
- In-app event buses: Laravel events with queued listeners, Spring's
ApplicationEventPublisher, MediatR notifications, Node.js event emitters with job queues. - Webhooks: an external publisher (Stripe, GitHub) notifies subscribers by topic.
Key points
- Pub/Sub puts a broker and named topics between publishers and subscribers; neither knows the other.
- Delivery is usually asynchronous, which isolates slow or failing consumers.
- Observer is the in-process, synchronous cousin; Pub/Sub scales across processes and services.
- Expect eventual consistency and duplicates: make subscribers idempotent and keep dead-letter queues.
- Further reading: Wikipedia: Publish-subscribe pattern.
Practice questions
What is the difference between Observer and Pub/Sub?
Answer
In Observer, the subject holds references to its observers and usually calls them synchronously in the same process. In Pub/Sub, publishers and subscribers only know a broker and topic names; delivery is typically asynchronous and can cross processes, with failures isolated by the broker.
Why must Pub/Sub subscribers usually be idempotent?
Answer
Most brokers deliver at least once: after timeouts or retries, the same message can arrive twice. An idempotent subscriber (one that records processed message IDs or uses conditional writes) produces the same result when processing a duplicate.
What is a dead-letter queue?
Answer
A place where messages go after a subscriber has failed to process them a set number of times. It stops a poison message from blocking or retrying forever and keeps it for inspection and replay.
Open in the app to mark lessons complete, take notes and track your progress.