Publish-Subscribe System

A message broker that lets publishers and subscribers never know about each other - only about a topic between them. It's the Observer pattern with a name change and a broker doing the introducing.

Requirements

Functional

A publisher publishes a message to a named topic.
A subscriber subscribes to a topic and receives every message published to it after it subscribed.
A subscriber can unsubscribe from a topic.
A topic can have many subscribers; a subscriber can be subscribed to many topics.

Non-functional

Delivery is at-least-once: a subscriber may see the same message twice on a retry, but must never silently miss one because a callback threw.
A slow or failing subscriber must not block delivery to the other subscribers of the same topic.

Design

Broker is the only thing publishers and subscribers ever talk to. A Topic holds its own subscriber list and does its own fan-out; the broker's job is just routing a publish to the right topic and creating topics on first use.

1A subscriber registers interest in a topic by name - it never talks to a Topic object directly.
2The broker creates the topic on first reference, for either a publisher or a subscriber.
3The publisher only knows a topic name and a message - not who, or how many, are listening.
4The broker forwards the publish to the matching topic and steps out of the way.
5The topic calls every subscriber in turn; one subscriber throwing does not stop the rest.

At-least-once delivery means Topic.publish treats one subscriber's exception as that subscriber's problem - it catches, logs, and keeps delivering to the rest, rather than letting one bad callback take down the whole fan-out.

Class diagram

implementsuses

Broker owns Topics; each Topic fans a Message out to its own Subscribers, independent of every other topic.

Code

import java.time.Instant;
import java.util.*;
 
class Message {
    final String id;
    final String payload;
    final Instant publishedAt;
 
    Message(String id, String payload) {
        this.id = id;
        this.payload = payload;
        this.publishedAt = Instant.now();
    }
}
 
interface Subscriber {
    void onMessage(Message message);
}
 
class LoggingSubscriber implements Subscriber {
    private final String name;
 
    LoggingSubscriber(String name) {
        this.name = name;
    }
 
    public void onMessage(Message message) {
        System.out.println(name + " received " + message.id + ": " + message.payload);
    }
}
 
class Topic {
    final String name;
    private final List<Subscriber> subscribers = new ArrayList<>();
 
    Topic(String name) {
        this.name = name;
    }
 
    void addSubscriber(Subscriber sub) {
        subscribers.add(sub);
    }
 
    void removeSubscriber(Subscriber sub) {
        subscribers.remove(sub);
    }
 
    void publish(Message message) {
        for (Subscriber sub : subscribers) {
            try {
                sub.onMessage(message);
            } catch (RuntimeException e) {
                System.out.println("Subscriber failed on " + message.id + ": " + e.getMessage());
            }
        }
    }
}
 
class Broker {
    private final Map<String, Topic> topics = new HashMap<>();
 
    private Topic getOrCreateTopic(String name) {
        return topics.computeIfAbsent(name, Topic::new);
    }
 
    void subscribe(String topicName, Subscriber sub) {
        getOrCreateTopic(topicName).addSubscriber(sub);
    }
 
    void unsubscribe(String topicName, Subscriber sub) {
        Topic topic = topics.get(topicName);
        if (topic != null) topic.removeSubscriber(sub);
    }
 
    void publish(String topicName, Message message) {
        getOrCreateTopic(topicName).publish(message);
    }
}

Design decisions

Subscriber is a callback interface, not a class the broker inspects. The broker never needs to know what a subscriber does with a message, only that it can be handed one. That's the whole reason this looks like Observer: the subject (Topic) is decoupled from every observer's implementation.
Topics are created lazily by the broker, not pre-registered. A publisher or subscriber referencing a topic name that doesn't exist yet shouldn't be an error case to special-case at every call site - getOrCreateTopic makes "topic doesn't exist" and "topic exists" the same code path.
Fan-out failure isolation lives inside Topic.publish, not in the broker. Each subscriber call is wrapped individually so one throwing subscriber can't stop the loop before the rest have been notified - that's what "at-least-once, not at-most-once" means in code.
What's missing for a real system: this broker delivers in-process and synchronously; a production version would persist messages to a durable log before acknowledging the publisher, and would give each subscriber its own offset/cursor so a restarted subscriber resumes instead of missing everything published while it was down.

Common follow-ups

What happens if a subscriber is slow or hangs during onMessage? Topic.publish calls subscribers synchronously in a loop, so a hanging subscriber blocks delivery to every subscriber after it on that topic - the fix is async per-subscriber dispatch (a queue or thread per subscriber) or a timeout around each call, neither of which this page's synchronous broker implements.
How do you get exactly-once or at-most-once instead of at-least-once? At-most-once falls out of removing the per-subscriber try/catch and simply skipping a failed subscriber without redelivery, accepting drops. True exactly-once needs subscriber-side deduplication (an idempotency key checked before processing), since Broker alone cannot guarantee a message is applied exactly once on the receiving end.
How would you add topic partitioning? Topic would hold multiple internal partitions, each with its own subscriber offset, and Broker.publish would hash the message key to pick one. Subscriber's interface doesn't change - only Topic's internal fan-out gains a routing step before it reaches the per-partition subscriber loop.
A subscriber unsubscribes from inside its own onMessage mid-publish - what breaks? Topic.publish iterates its subscriber list directly, so removing an entry during that iteration risks skipping or duplicating a subscriber (or throwing, depending on the language). A safer implementation iterates over a snapshot copy of the list, not the live one.

Check yourself

Question 1 of 4

Why is Subscriber a callback interface the broker never inspects, rather than the broker holding logic per subscriber type?

Requirements​

Design​

Class diagram​

Code​

Design decisions​

Common follow-ups​