Overview

A command consumer group turns a single command handler into a scalable worker pool: run multiple identical instances subscribed with the same group name, and the broker load-balances each incoming command to exactly one member instead of broadcasting it to all of them. This is how you add capacity to handle a growing command volume — start more workers in the same group — without changing anything on the caller's side.

Every subscriber sets the same group() alongside channel() on its CommandsSubscription, then calls subscribeToCommands(); the broker tracks membership and picks one live member per command. sendCommand() on the caller side is unaware groups exist — it just blocks for a response, which comes back from whichever handler happened to process it.

Gotchas: group membership is scoped per channel — subscribers on the same channel with different group names each get their own full copy of every command (fan-out), which looks like a bug when you expected load-balancing. A slow handler still holds up the caller's timeout, since only one worker is ever picked. And if every member of the group is offline when a command arrives, the send simply fails or times out — commands aren't queued or replayed for a group that has no active listener.

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

ConsumerGroupExample.java

package io.kubemq.example.commands;

import io.kubemq.sdk.cq.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerGroupExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-commands-consumer-group-client";
    private static final String CHANNEL = "java-commands.consumer-group";

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

        // Subscribe multiple handlers in a consumer group (load-balanced)
        String group = "command-handlers";
        int numWorkers = 3;
        AtomicInteger[] counts = new AtomicInteger[numWorkers];
        CommandsSubscription[] subs = new CommandsSubscription[numWorkers];

        for (int i = 0; i < numWorkers; i++) {
            final int id = i + 1;
            counts[i] = new AtomicInteger(0);
            final AtomicInteger counter = counts[i];

            subs[i] = CommandsSubscription.builder()
                    .channel(CHANNEL).group(group)
                    .onReceiveCommandCallback(cmd -> {
                        counter.incrementAndGet();
                        client.sendResponseMessage(CommandResponseMessage.builder()
                                .commandReceived(cmd).isExecuted(true).build());
                    })
                    .onErrorCallback(err -> {}).build();
            // Subscribe each worker to the consumer group
            client.subscribeToCommands(subs[i]);
        }
        Thread.sleep(500);

        // Send commands to the group (distributed across workers)
        System.out.println("Sending 9 commands to group '" + group + "'...\n");
        for (int i = 1; i <= 9; i++) {
            try { client.sendCommand(CommandMessage.builder()
                    .channel(CHANNEL).body(("Cmd #" + i).getBytes()).timeoutInSeconds(10).build()); }
            catch (Exception e) { /* handle */ }
        }

        System.out.println("Distribution:");
        for (int i = 0; i < numWorkers; i++) {
            System.out.println("  Worker " + (i + 1) + ": " + counts[i].get());
        }

        // Clean up resources
        for (CommandsSubscription s : subs) { s.cancel(); }
        client.deleteCommandsChannel(CHANNEL);
        client.close();
    }
}

How It Works

Three CommandsSubscription objects share the same group("command-handlers") on the same channel; KubeMQ delivers each incoming command to exactly one handler in the group (load-balanced), not all three.
Each handler increments its own AtomicInteger counter and sends isExecuted(true) back; the distribution printed at the end shows how the broker spread 9 commands across the 3 workers.
sendCommand() blocks until the group returns a response, so the loop completes only after all 9 commands are acknowledged.
In production each handler would run in a separate JVM or container; the group string is the only coupling between them and the channel.

Command Group

Overview

Prerequisites

Code

How It Works

On this page

Command Group

Overview

Prerequisites

Code

How It Works

Related

On this page