Skip to main content

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.

PublisherBrokerTopicSubscribersubscribe("orders", handler)1getOrCreateTopic("orders")2publish("orders", message)3publish(message)4onMessage(message)5
  1. 1A subscriber registers interest in a topic by name - it never talks to a Topic object directly.
  2. 2The broker creates the topic on first reference, for either a publisher or a subscriber.
  3. 3The publisher only knows a topic name and a message - not who, or how many, are listening.
  4. 4The broker forwards the publish to the matching topic and steps out of the way.
  5. 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​

«interface»Subscriber+ onMessage(message): voidBroker- topics: Map<string, Topic>+ publish(topicName, message): void+ subscribe(topicName, sub): void+ unsubscribe(topicName, sub): voidTopic- name: string- subscribers: List<Subscriber>+ publish(message): void+ addSubscriber(sub): void+ removeSubscriber(sub): voidMessage- id: string- payload: string- publishedAt: datetimeLoggingSubscriber+ onMessage(message): void
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.
0%0 of 122 pages studied