Design an in-process pub/sub event bus

01The design, in the six-step framework

1. Requirements and use cases

A component subscribes to a topic and receives every event published on it afterwards, in publish order. A publisher posts an event without knowing who listens and without waiting for them. A subscriber can cancel at any time, including from inside its own handler while a publish is in progress. A slow subscriber must not slow down the others. Shutdown stops delivery cleanly.

Out of scope: durability across restarts, delivery to other processes or machines (that is the managed pub/sub problem), replay from an offset, and wildcard topics.

Constraints: publishers and subscribers run on different threads; per-subscriber ordering must hold; memory must be bounded, so a subscriber that falls behind has to be handled by an explicit policy rather than unbounded growth; a handler that throws must not break delivery to anyone else.

2. Core entities and relationships

• EventBus: the mediator. It maps topics to subscribers and owns one bounded queue and one delivery thread per subscriber.
• Event: an immutable record with a global sequence number, so ordering is observable and tests can assert it.
• Subscriber: a protocol with a single on_event method; any object with that method qualifies, which keeps the interface small (interface segregation).
• Subscription: the token for cancelling; it carries the id so the bus never has to search by object identity.

The bus composes queues, threads and subscriptions and aggregates subscribers, which the application owns. Giving every subscriber its own queue is the structural decision that satisfies both "slow subscribers do not block others" and "per-subscriber order".

3. Interfaces and key methods

class Subscriber(Protocol):
    def on_event(self, topic: str, payload: object) -> None: ...

class Subscription:
    def cancel(self) -> None: ...          # safe to call during a publish

class EventBus:
    def subscribe(self, topic: str, subscriber: Subscriber) -> Subscription: ...
    def publish(self, topic: str, payload: object) -> int:
        """Enqueue for every current subscriber; returns how many were targeted."""
    def start(self, workers: int = 1) -> None: ...
    def shutdown(self, drain: bool = True) -> None: ...

Invariants: a subscriber id is in _subs if and only if it is in _queues (both change under the lock); a cancelled subscription's queue ends with exactly one poison pill; events for one subscriber are delivered in increasing seq.

4. Design patterns and the choices behind them

Observer with a queue in between

The plain observer calls every handler synchronously on the publisher's thread, so one slow handler delays the publisher and every other subscriber. Putting a bounded producer-consumer queue per subscriber decouples them: publish costs one enqueue per subscriber and returns. The rejected alternative, one shared queue with one pool of workers, cannot keep per-subscriber order without extra bookkeeping.

Snapshot the subscriber list under the lock

Unsubscribing during a publish is the classic concurrent-modification trap. publish copies the current targets under the lock and then releases it before enqueueing, so a handler that cancels its own subscription mid-fan-out cannot corrupt the iteration; it simply receives no further events after its queue's poison pill (see thread-safe collections).

A named policy for the slow subscriber

Bounded queues force a choice when full: block the publisher, drop the newest, or drop the oldest. The code drops the newest and counts it, because an in-process bus is usually for best-effort notifications and a blocked publisher is the worst outcome. Say the choice and its cost; an interviewer wants to hear that the policy is per-subscriber configuration, not a global constant.

Poison pill for shutdown

A None in the queue ends the delivery thread after it drains what was already enqueued, which is simpler and safer than a shared stop flag that the thread might check only after blocking on get.

5. Core flow and code

1. s = bus.subscribe("orders", handler): under the lock a new id is issued, the handler is registered for the topic, a queue is created and a delivery thread starts.
2. bus.publish("orders", order): the targets for the topic are copied and an Event with the next sequence number is built, both under the lock.
3. Outside the lock the event is put on each target's queue without blocking; a full queue increments that subscriber's drop counter.
4. The delivery thread for handler takes the event and calls on_event; an exception is swallowed and logged so the stream continues.
5. Inside its handler, the subscriber calls s.cancel(): it is removed from the maps and a poison pill is queued behind any events already waiting.
6. The delivery thread reaches the pill and exits; later publishes no longer see the id in the snapshot.

import collections, itertools, queue, threading
from dataclasses import dataclass
from typing import Protocol

class Subscriber(Protocol):
    def on_event(self, topic: str, payload: object) -> None: ...

@dataclass(frozen=True)
class Event:
    seq: int
    topic: str
    payload: object

class Subscription:
    def __init__(self, bus: "EventBus", topic: str, sub_id: int) -> None:
        self._bus, self.topic, self.sub_id = bus, topic, sub_id
        self.active = True

    def cancel(self) -> None:
        self._bus._unsubscribe(self)

class EventBus:
    """Each subscriber gets its own bounded queue so a slow one cannot stall the rest."""

    def __init__(self, queue_size: int = 1000) -> None:
        self._lock = threading.RLock()
        self._subs: dict[str, dict[int, Subscriber]] = collections.defaultdict(dict)
        self._queues: dict[int, queue.Queue[Event | None]] = {}
        self._ids = itertools.count(1)
        self._seq = itertools.count(1)
        self._queue_size = queue_size
        self._threads: list[threading.Thread] = []
        self._dropped: dict[int, int] = collections.defaultdict(int)

    def subscribe(self, topic: str, subscriber: Subscriber) -> Subscription:
        with self._lock:
            sub_id = next(self._ids)
            self._subs[topic][sub_id] = subscriber
            self._queues[sub_id] = queue.Queue(maxsize=self._queue_size)
            t = threading.Thread(target=self._deliver, args=(sub_id, subscriber), daemon=True)
            self._threads.append(t); t.start()
            return Subscription(self, topic, sub_id)

    def _unsubscribe(self, sub: Subscription) -> None:
        with self._lock:
            if not sub.active:
                return
            sub.active = False
            self._subs[sub.topic].pop(sub.sub_id, None)
            q = self._queues.pop(sub.sub_id, None)
        if q is not None:
            q.put(None)                       # poison pill ends the delivery thread

    def publish(self, topic: str, payload: object) -> int:
        with self._lock:                      # snapshot: unsubscribes during fan-out are safe
            targets = list(self._subs.get(topic, {}).items())
            event = Event(next(self._seq), topic, payload)
        for sub_id, _ in targets:
            q = self._queues.get(sub_id)
            if q is None:
                continue
            try:
                q.put_nowait(event)
            except queue.Full:
                self._dropped[sub_id] += 1    # policy: drop newest; see extensions
        return len(targets)

    def _deliver(self, sub_id: int, subscriber: Subscriber) -> None:
        q = self._queues[sub_id]
        while True:
            event = q.get()
            if event is None:
                return
            try:
                subscriber.on_event(event.topic, event.payload)
            except Exception:
                pass                          # one bad handler must not kill its own stream

    def start(self, workers: int = 1) -> None:
        pass                                  # delivery threads start on subscribe

    def shutdown(self, drain: bool = True) -> None:
        with self._lock:
            subs = [(t, sid) for t, d in self._subs.items() for sid in list(d)]
        for topic, sid in subs:
            self._unsubscribe(Subscription(self, topic, sid))
        if drain:
            for t in self._threads:
                t.join(timeout=1.0)

    def dropped(self, sub_id: int) -> int:
        return self._dropped.get(sub_id, 0)

6. Extensions, concurrency and what interviewers probe

• Push versus pull (the reported discussion point): the design is push into a queue, pull by the delivery thread. A pure pull API, poll(sub_id), is the same bus without the threads; offer it when subscribers already have a loop.
• Ordering across subscribers: only per-subscriber order is promised. Global order would need one queue and one consumer, which contradicts isolation; say so.
• Where the lock matters: registration and the snapshot in publish share one RLock, so a subscriber added during a publish either gets the event or does not, never half of a fan-out.
• Backpressure instead of drops: replace put_nowait with put(timeout) and surface the wait to the publisher; which one is right depends on whether events are notifications or commands.
• Test first: two subscribers on one topic, publish three events, assert each receives all three in order; a subscriber whose handler cancels itself on the first event, assert it receives exactly one; a subscriber with queue size 1 that never drains, assert the drop counter grows and the other subscriber is unaffected.
• Common mistake: iterating the live subscriber dict in publish, which raises "dictionary changed size during iteration" the first time a handler unsubscribes.

02Where this question was reported

• GitHub · awesome-low-level-design, Medium
• workat.tech · workat.tech machine coding 'Distributed Queue | Kafka' (SDE II/III)
• LLD Canvas · Push vs pull delivery decision
• thita.ai · thita.ai LLD sheet, advanced
• javarevisited.blogspot.com · javarevisited distributed message queue
• Codemia · Codemia Medium 'Event Emitter'