Event Bus and Observers in Java

An event bus lets one part of a program announce that something happened without knowing who cares. Order placed, file saved, sensor reading received: publishers emit, subscribers react, and neither side imports the other. It is the observer pattern with a broker in the middle, and it is a common low-level design prompt ("design a notification hub", "design a pub/sub library"). This article builds it in four steps, each one fixing a weakness of the previous one. For a cross-process version see design notification service.

Requirements

Functional

Subscribe a handler to a topic and get back a way to unsubscribe.
Publish an event to a topic; every current subscriber of that topic receives it.
Unsubscribe by id; unsubscribing twice is harmless.
Subscribers of other topics never see the event.

Non-functional

Delivery order to subscribers is deterministic (subscription order).
A handler may subscribe or unsubscribe while an event is being delivered.

Entities

classDiagram
    class EventBus {
        -nextId: long
        -topics: Map~String, List~Subscriber~~
        +subscribe(topic, handler) SubscriptionId
        +unsubscribe(id) boolean
        +publish(topic, event) int
    }
    class Subscriber {
        <<record>>
        id: SubscriptionId
        handler: Consumer~Event~
    }
    class SubscriptionId {
        <<record>>
        value: long
    }
    EventBus "1" *-- "*" Subscriber
    Subscriber --> SubscriptionId

The bus holds its subscribers in a map. A subscriber is an id plus a handler. Nothing in the subscriber points back to the bus, so the object graph has no cycle. Refer to subscriptions by id, not by the handler object: two subscriptions may use the same lambda, and an id names exactly one of them.

Step 1: callbacks and unsubscribe by id

A handler is a Consumer<Event>: a functional interface with one method, accept. A lambda or a method reference fits it, and subscribers usually need to capture a little state, so a lambda is lighter than a dedicated one-method class.

The bus below is single-threaded. subscribe returns a SubscriptionId record, and unsubscribe removes the entry with that id. TreeMap keeps topics sorted so any listing is stable. publish iterates over a copy of the subscriber list: a handler that subscribes to the same topic while the loop runs would otherwise cause a ConcurrentModificationException, because ArrayList detects structural changes made during iteration. The copy means a subscriber added during delivery receives the next event, not the current one.

Java

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.function.Consumer;

record SubscriptionId(long value) {}

record Event(String name, long amountCents) {}

record Subscriber(SubscriptionId id, Consumer<Event> handler) {}

final class EventBus {
    private long nextId;
    private final Map<String, List<Subscriber>> topics = new TreeMap<>();

    SubscriptionId subscribe(String topic, Consumer<Event> handler) {
        var id = new SubscriptionId(++nextId);
        topics.computeIfAbsent(topic, t -> new ArrayList<>()).add(new Subscriber(id, handler));
        return id;
    }

    boolean unsubscribe(SubscriptionId id) {
        boolean removed = false;
        for (var subscribers : topics.values()) {
            removed |= subscribers.removeIf(s -> s.id().equals(id));
        }
        return removed;
    }

    /** Returns how many subscribers received the event. */
    int publish(String topic, Event event) {
        var subscribers = topics.get(topic);
        if (subscribers == null) {
            return 0;
        }
        var snapshot = List.copyOf(subscribers);
        for (var subscriber : snapshot) {
            subscriber.handler().accept(event);
        }
        return snapshot.size();
    }
}

final class RunningTotal {
    private long total;

    void add(Event event) {
        total += event.amountCents();
        System.out.println("running total: " + total);
    }
}

public class Main {
    public static void main(String[] args) {
        var bus = new EventBus();

        var first = bus.subscribe("orders", e -> System.out.println("audit: " + e.name() + " " + e.amountCents()));
        bus.subscribe("orders", new RunningTotal()::add);

        System.out.println("delivered to " + bus.publish("orders", new Event("order-1", 1250)));

        System.out.println("unsubscribed first: " + bus.unsubscribe(first));
        System.out.println("unsubscribed again: " + bus.unsubscribe(first));

        System.out.println("delivered to " + bus.publish("orders", new Event("order-2", 500)));

        // A handler may subscribe during delivery; the new handler sees the next event.
        bus.subscribe("alerts", e -> bus.subscribe("alerts", later -> System.out.println("late: " + later.name())));
        System.out.println("delivered to " + bus.publish("alerts", new Event("a1", 0)));
        System.out.println("delivered to " + bus.publish("alerts", new Event("a2", 0)));
    }
}

Output

audit: order-1 1250
running total: 1250
delivered to 2
unsubscribed first: true
unsubscribed again: false
running total: 1750
delivered to 1
delivered to 1
late: a2
delivered to 2

The second subscriber is a method reference on a RunningTotal object, which keeps the running total in a field. A lambda cannot modify a local variable it captures, so mutable subscriber state lives in an object. Amounts are integer cents in a long, never floating point.

This version has three weaknesses. The topic is a free-form string, so a typo compiles and silently delivers to nobody. The payload is one fixed Event type. And unsubscribing needs the caller to keep the id and call the bus, which callers forget. The next steps fix these.

Step 2: scoped unsubscribe with `AutoCloseable`

Forgotten unsubscribes are the classic observer leak: the bus keeps a strong reference to every handler, so a handler that is no longer needed (and everything it captured) stays reachable until someone removes it. Java has no destructor that runs at a known moment, so the scoped form is AutoCloseable with try-with-resources. Subscription.close removes the handler, and calling it twice is harmless.

The Subscription holds a reference to the bus. The bus never holds a Subscription, only handlers, so there is no cycle. Even a cycle would be collectible by the garbage collector, so this design needs no weak reference. A bus could hold its handlers through WeakReference instead, but then a handler with no other reference disappears when the garbage collector decides, so delivery would depend on collection timing. This lesson does not use that design.

Java

import java.util.ArrayList;
import java.util.Map;
import java.util.TreeMap;
import java.util.function.Consumer;

@FunctionalInterface
interface Subscription extends AutoCloseable {
    @Override
    void close(); // no checked exception, so try-with-resources needs no catch
}

final class Bus {
    private long nextId;
    private final Map<Long, Consumer<String>> handlers = new TreeMap<>();

    Subscription subscribe(Consumer<String> handler) {
        long id = ++nextId;
        handlers.put(id, handler);
        return () -> handlers.remove(id);
    }

    void publish(String message) {
        for (var handler : new ArrayList<>(handlers.values())) {
            handler.accept(message);
        }
    }

    int size() {
        return handlers.size();
    }
}

public class Main {
    public static void main(String[] args) {
        var bus = new Bus();

        Subscription log = bus.subscribe(m -> System.out.println("log: " + m));
        try (Subscription temporary = bus.subscribe(m -> System.out.println("temporary: " + m))) {
            bus.publish("first");
            System.out.println("subscribers inside: " + bus.size());
        } // temporary.close() runs here and removes the handler

        bus.publish("second");
        System.out.println("subscribers after: " + bus.size());

        log.close();
        log.close(); // closing twice is harmless
        System.out.println("subscribers at end: " + bus.size());
    }
}

Output

log: first
temporary: first
subscribers inside: 2
log: second
subscribers after: 1
subscribers at end: 0

The ids come from a counter, so a TreeMap keyed by id delivers in subscription order. The handler of log is never closed in a scope: it lives as long as the bus, which is the right lifetime for a long-lived subscriber. A handle that nobody closes keeps its handler registered for the life of the bus.

Step 3: mailbox subscribers

Callbacks run on the publisher's thread, inside publish. A slow or throwing handler slows or breaks the publisher. The alternative is to give each subscriber a mailbox: the bus puts the event in a queue, and the subscriber takes it whenever it likes, possibly from another thread. Publisher and subscriber are decoupled in time. The queue is a LinkedBlockingQueue; see blocking queue for how a blocking queue works.

A queue has no built-in close, so the end of a stream is a message too. A sealed interface Delivery has two cases: Item carries an event and End says no more events are coming. A switch over a sealed type must cover both cases, so a reader cannot forget the end marker.

Java

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

sealed interface Delivery<E> permits Item, End {}

record Item<E>(E value) implements Delivery<E> {}

record End<E>() implements Delivery<E> {}

/** One consumer thread reads a mailbox. */
final class Mailbox<E> {
    private final BlockingQueue<Delivery<E>> queue = new LinkedBlockingQueue<>();
    private boolean ended; // touched only by the reading thread

    void put(Delivery<E> delivery) {
        queue.add(delivery);
    }

    /** Blocks for the next event. Returns empty after the bus ends this mailbox. */
    Optional<E> take() throws InterruptedException {
        if (ended) {
            return Optional.empty();
        }
        return switch (queue.take()) {
            case Item<E>(var value) -> Optional.of(value);
            case End<E> end -> {
                ended = true;
                yield Optional.empty();
            }
        };
    }
}

record MailboxSubscription<E>(long id, Mailbox<E> mailbox) {}

/** Used from one publishing thread. Readers only touch their own mailbox. */
final class ChannelBus<E> {
    private long nextId;
    private final Map<Long, Mailbox<E>> subscribers = new TreeMap<>();

    MailboxSubscription<E> subscribe() {
        var subscription = new MailboxSubscription<>(++nextId, new Mailbox<E>());
        subscribers.put(subscription.id(), subscription.mailbox());
        return subscription;
    }

    /** Ends the mailbox, so the reader's loop finishes after the events already queued. */
    boolean unsubscribe(long id) {
        var mailbox = subscribers.remove(id);
        if (mailbox == null) {
            return false;
        }
        mailbox.put(new End<>());
        return true;
    }

    /** Shares the same event object with every subscriber, so events must be immutable. */
    int publish(E event) {
        for (var mailbox : subscribers.values()) {
            mailbox.put(new Item<>(event));
        }
        return subscribers.size();
    }

    void close() {
        for (var id : List.copyOf(subscribers.keySet())) {
            unsubscribe(id);
        }
    }
}

public class Main {
    static List<String> drain(Mailbox<String> mailbox) throws InterruptedException {
        var received = new ArrayList<String>();
        for (var event = mailbox.take(); event.isPresent(); event = mailbox.take()) {
            received.add(event.get());
        }
        return received;
    }

    public static void main(String[] args) throws Exception {
        var bus = new ChannelBus<String>();
        var a = bus.subscribe();
        var b = bus.subscribe();

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Future<List<String>> readerA = executor.submit(() -> drain(a.mailbox()));
            Future<List<String>> readerB = executor.submit(() -> drain(b.mailbox()));

            System.out.println("live after publish: " + bus.publish("started"));
            bus.publish("running");
            System.out.println("b unsubscribed: " + bus.unsubscribe(b.id()));
            bus.publish("stopped");
            bus.close(); // ends the remaining mailboxes

            System.out.println("a got " + readerA.get());
            System.out.println("b got " + readerB.get());
        }
    }
}

Output

live after publish: 2
b unsubscribed: true
a got [started, running, stopped]
b got [started, running]

Four details carry the design. Each mailbox holds one Delivery per event, so a subscriber that never reads fills memory: an unbounded queue grows until memory runs out. A bounded mailbox needs a policy for the full case (block, drop the newest, drop the oldest), listed under extensions. The bus shares one event object among all mailboxes, which is safe only when the event is immutable, such as a record of immutable fields. The End marker is queued after every event already published, so subscriber B sees exactly the two events published before it unsubscribed, whatever the reader's speed. And a mailbox has one reading thread, which is why ended needs no lock.

Step 4: typed topics and a thread-safe bus

Strings and one shared event type do not scale. A typed topic ties a payload type to a topic value, so publishing the wrong payload is a compile error and subscribers need no casts. A Topic<E> carries a name and a Class<E> object, the runtime token for the payload type.

Java erases generic type arguments at runtime, so the bus cannot keep a different List<Handler<E>> for each E in one map without an unchecked cast. The bus stores each handler as Consumer<Object> and wraps the typed handler in a lambda that calls payloadType.cast(event). The cast cannot fail while the topic name is used with one payload type. A name reused with another type is detected by comparing the Class objects, and the bus throws TopicTypeMismatchException. The exception is unchecked because it reports a programming error in the caller, the same category as IllegalArgumentException.

The bus is thread-safe and calls handlers outside the lock: it copies the handler list under the lock, releases the lock, then calls the handlers. That lets a handler subscribe or unsubscribe during delivery. Without it, a handler that waits for another thread that wants to subscribe would deadlock, because the handler holds the lock while it waits. This is the rule from thread-safe components: never call foreign code while holding a lock.

Java

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Consumer;

/** A topic value carries its payload type. */
record Topic<E>(String name, Class<E> payloadType) {
    Topic {
        Objects.requireNonNull(name);
        Objects.requireNonNull(payloadType);
    }
}

record SubscriptionId(long value) {}

final class TopicTypeMismatchException extends RuntimeException {
    TopicTypeMismatchException(String topic, Class<?> existing, Class<?> requested) {
        super("topic '%s' carries %s, not %s".formatted(topic, existing.getSimpleName(), requested.getSimpleName()));
    }
}

final class EventBus {
    private record Entry(SubscriptionId id, Consumer<Object> handler) {}

    private static final class Channel {
        final Class<?> payloadType;
        final List<Entry> entries = new ArrayList<>();

        Channel(Class<?> payloadType) {
            this.payloadType = payloadType;
        }
    }

    private final Object lock = new Object();
    private long nextId; // guarded by lock
    private final Map<String, Channel> topics = new TreeMap<>(); // guarded by lock

    <E> SubscriptionId subscribe(Topic<E> topic, Consumer<? super E> handler) {
        Objects.requireNonNull(handler);
        synchronized (lock) {
            var channel = topics.computeIfAbsent(topic.name(), name -> new Channel(topic.payloadType()));
            requireSameType(channel, topic);
            var id = new SubscriptionId(++nextId);
            channel.entries.add(new Entry(id, event -> handler.accept(topic.payloadType().cast(event))));
            return id;
        }
    }

    boolean unsubscribe(SubscriptionId id) {
        synchronized (lock) {
            boolean removed = false;
            for (var channel : topics.values()) {
                removed |= channel.entries.removeIf(entry -> entry.id().equals(id));
            }
            return removed;
        }
    }

    <E> int publish(Topic<E> topic, E event) {
        Objects.requireNonNull(event);
        List<Entry> snapshot;
        synchronized (lock) {
            var channel = topics.get(topic.name());
            if (channel == null) {
                return 0;
            }
            requireSameType(channel, topic);
            snapshot = List.copyOf(channel.entries); // copy the handlers, then release the lock
        }
        for (var entry : snapshot) {
            entry.handler().accept(event);
        }
        return snapshot.size();
    }

    private static void requireSameType(Channel channel, Topic<?> topic) {
        if (!channel.payloadType.equals(topic.payloadType())) {
            throw new TopicTypeMismatchException(topic.name(), channel.payloadType, topic.payloadType());
        }
    }
}

record OrderPlaced(int orderId, long cents) {}

public class Main {
    static final Topic<OrderPlaced> ORDERS = new Topic<>("orders", OrderPlaced.class);
    static final Topic<String> ALERTS = new Topic<>("alerts", String.class);

    public static void main(String[] args) throws Exception {
        var bus = new EventBus();
        var revenue = new LongAdder();

        var id = bus.subscribe(ORDERS, order -> revenue.add(order.cents()));
        bus.subscribe(ALERTS, message -> System.out.println("alert: " + message));

        // Four threads publish concurrently; the sum does not depend on the schedule.
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<Integer>> publishers = new ArrayList<>();
            for (int n = 1; n <= 4; n++) {
                int orderId = n;
                publishers.add(executor.submit(() -> bus.publish(ORDERS, new OrderPlaced(orderId, 250L * orderId))));
            }
            for (var publisher : publishers) {
                publisher.get();
            }
        }
        System.out.println("revenue: " + revenue.sum());

        // A handler may unsubscribe itself during delivery.
        var once = new AtomicReference<SubscriptionId>();
        once.set(bus.subscribe(ALERTS, message -> {
            System.out.println("once: " + message);
            bus.unsubscribe(once.get());
        }));
        System.out.println("delivered: " + bus.publish(ALERTS, "disk full"));
        System.out.println("delivered: " + bus.publish(ALERTS, "disk full again"));

        System.out.println("removed: " + bus.unsubscribe(id));
        System.out.println("delivered: " + bus.publish(ORDERS, new OrderPlaced(9, 1)));

        // Reusing a name with a different payload type is reported, not ignored.
        var wrong = new Topic<>("orders", Integer.class);
        try {
            bus.subscribe(wrong, n -> {});
        } catch (TopicTypeMismatchException e) {
            System.out.println(e.getMessage());
        }
    }
}

Output

revenue: 2500
alert: disk full
once: disk full
delivered: 2
alert: disk full again
delivered: 1
removed: true
delivered: 0
topic 'orders' carries OrderPlaced, not Integer

The compiler now enforces the contract. bus.publish(ORDERS, "text") does not compile because ORDERS is a Topic<OrderPlaced>. The only runtime check left is the name collision at the end, which is possible because two Topic values can share a name. Declaring one constant per topic in one place removes that in practice.

The handler that unsubscribes itself still receives the current event, because it was in the copied list, and receives nothing afterwards. Handlers run on the publisher's thread, in subscription order for one publisher. Two publishers on the same topic interleave in an unspecified order, which is why the revenue check sums amounts instead of listing them.

Design decisions

Decision	Choice	Reason
Handler type	`Consumer<E>` lambda or method reference	subscribers are small; mutable state lives in an object the method reference points to
Unsubscribe	id plus optional `AutoCloseable` handle	id for explicit control, try-with-resources to prevent leaks
Delivery	synchronous callbacks, or a mailbox per subscriber	callbacks for ordering, mailboxes for isolation
Topics	typed `Topic<E>` with a `Class<E>` token	wrong payload is a compile error
Locking	lock to read the list, call handlers unlocked	handlers may re-enter the bus
Event objects	immutable records	one object is shared by every subscriber and thread

Extensions

Bounded mailboxes. Use ArrayBlockingQueue per subscriber with offer, and decide the policy when it returns false: drop the newest, drop the oldest, or disconnect the subscriber. See blocking queue.
Handler failures. A handler that throws propagates to the publisher and skips the remaining handlers in this version. Catch RuntimeException around each call and report it to an error listener to isolate subscribers.
Testing. Subscribe a lambda that adds to a list and assert on the list; inject nothing else. See testability and dependency injection.

Interview points

The observer pattern with a broker: publishers and subscribers share only the topic and the event type.
Mutating a list while iterating it throws ConcurrentModificationException; iterate over a copy, and state the snapshot semantics (late subscribers see the next event).
A strong reference from the bus to every handler is the lapsed-listener leak; try-with-resources or an explicit close prevents it.
Calling handlers under the lock invites deadlock and blocks every other publisher; copy under the lock, call outside.
A queue-based subscriber needs an end marker, a bound, and a policy for a full queue.
Type erasure: a Class<E> token lets the bus cast and compare payload types at runtime.

Summary

An event bus is a broker holding subscribers by id; publishers and subscribers never reference each other.
Lambdas are the natural handler type; SubscriptionId is a record and unsubscribe by id is idempotent.
A Subscription that implements AutoCloseable removes forgotten handlers in a try-with-resources scope, with no cycle between handle and bus.
Mailbox subscribers decouple timing and threads; the end of a stream is a message, and an unbounded mailbox grows with a slow reader.
Typed Topic<E> moves payload checking to compile time and leaves one runtime check for name collisions.
In a shared bus, copy the handler list under the lock and call handlers after releasing it, so handlers can re-enter the bus.
Keep delivery order deterministic (subscription order) so tests and behavior are stable. See also thread-safe components.

Requirements

Entities

Step 1: callbacks and unsubscribe by id

Step 2: scoped unsubscribe with AutoCloseable

Step 3: mailbox subscribers

Step 4: typed topics and a thread-safe bus

Design decisions

Extensions

Interview points

Summary

Step 2: scoped unsubscribe with `AutoCloseable`