Courses › Low-Level Design › Problem Walkthroughs

Design an In-Memory Pub/Sub Broker

Per-subscriber queues, ordering, backpressure, retries, dead letters and replay.

"Design an in-memory pub/sub message broker" looks like a simple Observer exercise until the follow-ups arrive. What if a subscriber is slow? What if one throws? Can subscribers replay old messages? It's a good showcase for clean interfaces, delivery semantics and concurrency in a single question. Logging frameworks and notification services are built on the same ideas.

Requirements

• Publishers send messages to a topic. Subscribers subscribe to one or more topics.
• Every subscriber of a topic receives every message published to it, in order.
• A slow or failing subscriber must not block the publisher or the other subscribers.
• Subscribers can unsubscribe.
• Follow-ups: replaying from an offset, filters and wildcard topics, retries and dead letters.

Two designs

Push (Observer): the broker calls subscriber.on_message(msg) for each subscriber. It's simple, but a slow callback slows down the publisher unless delivery happens on another thread.

Log + cursors (Kafka-style): each topic is an append-only list of messages, and each subscriber tracks its own offset in it. Publishing is just an append. Subscribers read at their own pace, and replaying means resetting the offset.

Start with push and per-subscriber queues, which is what most interviewers expect. Then mention the log design for replay.

Model

Message          id, topic, payload, timestamp
Subscriber       (interface) id, on_message(msg)
Subscription     subscriber, queue, worker thread, filter
Topic            name, subscriptions
Broker           create_topic, subscribe(topic, sub), unsubscribe, publish(topic, payload), shutdown

Isolating subscribers with their own queue

Each subscription gets a bounded queue and a worker thread. Publishing puts the message on each subscriber's queue and returns immediately. A slow subscriber only fills up its own queue.

import itertools
import queue
import threading
import time
from dataclasses import dataclass, field

_STOP = object()


@dataclass(frozen=True)
class Message:
    id: int
    topic: str
    payload: object
    ts: float = field(default_factory=time.time)


class Subscription:
    def __init__(self, subscriber, capacity=1000, max_retries=3, dead_letters=None):
        self.subscriber = subscriber
        self.queue = queue.Queue(maxsize=capacity)
        self.max_retries = max_retries
        self.dead_letters = dead_letters
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()

    def deliver(self, msg):
        try:
            self.queue.put_nowait(msg)
            return True
        except queue.Full:
            return False                      # backpressure policy: caller decides

    def _run(self):
        while True:
            msg = self.queue.get()
            if msg is _STOP:
                return
            for attempt in range(self.max_retries + 1):
                try:
                    self.subscriber.on_message(msg)
                    break
                except Exception:
                    if attempt == self.max_retries and self.dead_letters is not None:
                        self.dead_letters.append((self.subscriber.id, msg))

    def stop(self):
        self.queue.put(_STOP)                 # drains queued messages first, then exits
        self.thread.join()


class Broker:
    def __init__(self):
        self._topics = {}                     # topic -> {subscriber_id: Subscription}
        self._lock = threading.Lock()
        self._ids = itertools.count(1)
        self.dead_letters = []
        self.dropped = 0

    def subscribe(self, topic, subscriber):
        with self._lock:
            subs = self._topics.setdefault(topic, {})
            if subscriber.id in subs:
                raise ValueError("already subscribed")
            subs[subscriber.id] = Subscription(subscriber, dead_letters=self.dead_letters)

    def unsubscribe(self, topic, subscriber_id):
        with self._lock:
            sub = self._topics.get(topic, {}).pop(subscriber_id, None)
        if sub:
            sub.stop()                        # outside the lock: may wait for the queue to drain

    def publish(self, topic, payload):
        msg = Message(next(self._ids), topic, payload)
        with self._lock:
            subs = list(self._topics.get(topic, {}).values())   # snapshot, then release the lock
        for sub in subs:
            if not sub.deliver(msg):
                self.dropped += 1
        return msg.id

    def shutdown(self):
        with self._lock:
            subs = [s for t in self._topics.values() for s in t.values()]
            self._topics.clear()
        for s in subs:
            s.stop()

class Printer:
    def __init__(self, id): self.id, self.seen = id, []
    def on_message(self, msg): self.seen.append(msg.payload)

broker = Broker()
a = Printer("a")
broker.subscribe("orders", a)
for i in range(3):
    broker.publish("orders", {"order": i})
broker.shutdown()
assert [m["order"] for m in a.seen] == [0, 1, 2]

Decisions to explain

• Ordering: one queue and one worker per subscription keeps messages in publish order for that subscriber. A shared thread pool would break ordering unless messages are partitioned by key.
• Snapshot the subscriber list under the lock, then deliver outside it. Never call user code while holding the broker's lock. A callback that calls subscribe would deadlock.
• Backpressure policy for full queues: drop the newest message (what this code does, with a counter), drop the oldest, block the publisher, or disconnect the slow subscriber. Name the trade-off and choose based on the use case: telemetry can drop messages, payments can't.
• Failures: retry a limited number of times, then send the message to a dead-letter list. One bad subscriber never affects the others.
• Delivery guarantee: this is *at most once*. Messages are lost if the process crashes. *At least once* needs a durable log plus acknowledged offsets, which means subscribers must be idempotent. See Message Queues.

[!INTERVIEW] Push vs. pull "The push design isolates subscribers with queues. If we need replay or very slow consumers, I'd switch to a per-topic log where each subscriber keeps an offset. Publishing becomes an append, and replay is just resetting the offset." Comparing both designs is usually what separates a hire from a strong hire on this question.

Follow-ups

• Wildcard topics (orders.*): match topics with a trie over the dot-separated segments, as in Design an HTTP Router.
• Message filters: a predicate on the subscription, checked before enqueueing.
• Retention: in the log design, drop messages older than T, or below the minimum offset across all subscribers.
• Logging framework: the same shape. Loggers publish records, and appenders (console, file, remote) are subscribers with level filters, sometimes chained (Chain of Responsibility).

[!TIP] Practice In-Memory Pub/Sub Broker · Logging Framework · Notification Service · Bounded Blocking Queue