Overview

Fan-out delivery lets several independent consumers each get their own copy of every event published on a channel — the pattern behind broadcasting a notification to every connected service or feeding the same stream to a cache invalidator and a metrics collector at once. Reach for it whenever multiple, unrelated pieces of code all need to react to the same event, rather than compete for it.

It works by calling subscribeToEvents more than once for the same channel with separate EventsSubscription objects that leave .group() unset. Each subscription opens its own stream, and the broker treats every subscriber with no group as broadcast: publishing one event delivers it to every open stream — the opposite of a consumer group, where subscribers sharing a group split events among themselves for load balancing.

Gotchas: Events pub/sub has no durability — a subscriber that hasn't finished subscribing yet, or that disconnects, simply misses events published in that window; there's no redelivery. Setting .group() on one subscriber on the same channel silently turns broadcast into load-balancing for it. And because delivery runs on separate callback threads, shared state your callbacks touch needs its own synchronization.

Prerequisites

KubeMQ server running on localhost:50000
Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

MultipleSubscribersExample.java

package io.kubemq.example.events;

import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Multiple Subscribers Example
 *
 * Demonstrates multiple independent subscribers (no group) each receiving
 * all events -- broadcast mode.
 */
public class MultipleSubscribersExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-events-multiple-subscribers-client";
    private static final String CHANNEL = "java-events.multiple-subscribers";

    public static void main(String[] args) throws InterruptedException {
        // Create a client connected to the KubeMQ server
        PubSubClient client = PubSubClient.builder()
                .address(ADDRESS)
                .clientId(CLIENT_ID)
                .build();

        // Verify connection to the server
        ServerInfo info = client.ping();
        System.out.println("Connected to: " + info.getHost());
        // Create the events channel
        client.createEventsChannel(CHANNEL);

        int numMessages = 3;
        AtomicInteger sub1Count = new AtomicInteger(0);
        AtomicInteger sub2Count = new AtomicInteger(0);
        CountDownLatch latch = new CountDownLatch(numMessages * 2);

        EventsSubscription subscription1 = EventsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveEventCallback(event -> {
                    sub1Count.incrementAndGet();
                    System.out.println("  Subscriber A: " + new String(event.getBody()));
                    latch.countDown();
                })
                .onErrorCallback(err -> {})
                .build();

        EventsSubscription subscription2 = EventsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveEventCallback(event -> {
                    sub2Count.incrementAndGet();
                    System.out.println("  Subscriber B: " + new String(event.getBody()));
                    latch.countDown();
                })
                .onErrorCallback(err -> {})
                .build();

        // Subscribe two independent subscribers (no group = broadcast to all)
        client.subscribeToEvents(subscription1);
        client.subscribeToEvents(subscription2);
        System.out.println("Two independent subscribers created (broadcast mode).\n");

        // Wait for subscribers to be ready
        Thread.sleep(500);

        // Send event messages (each subscriber receives all)
        for (int i = 1; i <= numMessages; i++) {
            client.publishEvent(EventMessage.builder()
                    .id("msg-" + i)
                    .channel(CHANNEL)
                    .body(("Broadcast message #" + i).getBytes())
                    .build());
            Thread.sleep(100);
        }

        // Wait for both subscribers to receive all messages
        latch.await(5, TimeUnit.SECONDS);

        // Handle response: print counts per subscriber
        System.out.println("\nResults:");
        System.out.println("  Subscriber A received: " + sub1Count.get());
        System.out.println("  Subscriber B received: " + sub2Count.get());
        System.out.println("  Both received all " + numMessages + " messages (broadcast).");

        // Clean up resources
        subscription1.cancel();
        subscription2.cancel();
        client.deleteEventsChannel(CHANNEL);
        client.close();
        System.out.println("\nMultiple subscribers example completed.");
    }
}

How It Works

Two independent EventsSubscription objects are registered without a .group(), so both receive all published events (broadcast mode).
CountDownLatch(numMessages * 2) counts down once per delivery; it reaches zero only when each of the two subscribers has received all three messages.
Each subscriber callback uses its own AtomicInteger counter so the final summary can confirm that both received the full set independently.

Multiple Subscribers

Overview

Prerequisites

Code

How It Works

Related

On this page