# Command Group (/sdks/java/how-to/rpc/command-group)



## Overview [#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 [#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](/sdks/java))

## Code [#code]

```java title="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 [#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.

## Related [#related]

* [Pattern overview](/learn/rpc/getting-started)
* [Java SDK Reference](/sdks/java/reference/rpc)
* [Send Command](/sdks/java/tutorials/command-send)
* [Handle Command](/sdks/java/how-to/rpc/command-handle)