Design a Notification/Observer-Based Pub-Sub

Core classes

interface Subscriber<T> { void onEvent(T event); }

class Topic<T> {
    private final List<Subscriber<T>> subscribers = new CopyOnWriteArrayList<>(); // see below for why this, not ArrayList
    void subscribe(Subscriber<T> subscriber) { subscribers.add(subscriber); }
    void unsubscribe(Subscriber<T> subscriber) { subscribers.remove(subscriber); }
    List<Subscriber<T>> getSubscribers() { return subscribers; }
}

class EventBus {
    private final Map<String, Topic<Object>> topics = new ConcurrentHashMap<>();
    Topic<Object> topic(String name) { return topics.computeIfAbsent(name, k -> new Topic<>()); }
}

CopyOnWriteArrayList is a deliberate choice here, not an arbitrary one: subscriber lists are read far more often than written (an event publish iterates every subscriber; subscribe/unsubscribe happen comparatively rarely) — exactly the read-heavy, write-rare profile CopyOnWriteArrayList is built for (see ConcurrentHashMap & CopyOnWriteArrayList), avoiding the need to lock the whole list on every single publish.

Sync vs async delivery

class EventPublisher {
    void publishSync(Topic<Object> topic, Object event) {
        for (Subscriber<Object> s : topic.getSubscribers()) s.onEvent(event); // caller blocks until every subscriber finishes
    }

    void publishAsync(Topic<Object> topic, Object event, ExecutorService executor) {
        for (Subscriber<Object> s : topic.getSubscribers()) {
            executor.submit(() -> s.onEvent(event)); // fire-and-forget per subscriber, publisher returns immediately
        }
    }
}

Sync delivery guarantees every subscriber has processed the event before publish() returns — simpler to reason about, but one slow or failing subscriber blocks (or breaks) the publisher and every other subscriber's timing. Async delivery (dispatching each subscriber's callback onto an executor, see ExecutorService & Thread Pools) decouples the publisher's timing from any individual subscriber's processing time — closer to Inter-Service Communication Choices' fire-and-forget/pub-sub distinction, just applied in-process rather than across a network.

Avoiding the observer memory leak

class NotificationListener implements Subscriber<OrderEvent> {
    NotificationListener(Topic<OrderEvent> topic) {
        topic.subscribe(this); // subscribed...
    }
    void shutdown(Topic<OrderEvent> topic) {
        topic.unsubscribe(this); // ...MUST be explicitly unsubscribed, or this instance leaks forever
    }
}

Exactly the leak Observer Pattern warns about: a subscriber that's meant to be short-lived but never calls unsubscribe() stays referenced by the Topic's subscriber list indefinitely, preventing garbage collection even after nothing else in the application holds a reference to it. The fix pattern is the same too — either disciplined explicit unsubscription tied to a clear lifecycle hook (a @PreDestroy, a close() method), or storing subscribers as weak references so the GC can reclaim ones nothing else holds onto, accepting that a weakly-referenced subscriber might silently stop receiving events once collected rather than failing loudly.

Follow-up questions this topic invites — and their answers

Q: What happens to a subscriber's onEvent() exception during sync publish? A: In the naive loop shown, an exception in one subscriber's onEvent() would propagate and prevent later subscribers in the list from being notified — production implementations typically wrap each subscriber call in its own try/catch so one misbehaving subscriber can't break delivery to the rest, the same fix noted in Observer Pattern.

Q: Why ConcurrentHashMap for the topics map but CopyOnWriteArrayList for each topic's subscriber list? A: Different access patterns: topics themselves are created relatively rarely and looked up frequently by many threads (ConcurrentHashMap's general-purpose profile fits), while a single topic's subscriber list is specifically read-heavy/write-rare (CopyOnWriteArrayList's specific optimization target) — using the tool matched to each structure's actual usage pattern rather than one generic concurrent collection everywhere.

Q: How would you add delivery guarantees (at-least-once, retry) to the async path? A: This starts to become the Outbox Pattern / Saga Pattern territory — an in-memory EventBus with no persistence loses events on a crash; genuine delivery guarantees need the event durably stored before dispatch (an outbox table, or an actual message broker) rather than a fire-and-forget executor submission, which is exactly why this in-memory design is explicitly a simplified LLD exercise, not a production message broker (see also Design a Pub-Sub Message Broker for that distinction stated directly).

Q: Could this EventBus deadlock if a subscriber's onEvent() itself publishes to the same topic synchronously? A: With CopyOnWriteArrayList's iteration (which iterates over a snapshot, not the live list), a subscriber publishing back into the same topic during iteration won't corrupt the ongoing iteration or deadlock on the list itself — but it CAN cause unbounded recursive publish chains if not designed carefully, which is a logical/design risk to watch for independent of the underlying collection's thread-safety.