Command Group

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 instances of the process in the same group — without changing anything on the caller's side.

Every subscriber sets the same Group alongside Channel on its CommandsSubscription passed to SubscribeToCommandsAsync; the broker tracks membership and picks one live member per command. SendCommandResponseAsync is what a handler must call to complete the RPC — the broker uses the incoming command's RequestId and ReplyChannel to route the reply back to the original caller, whichever worker it came from.

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
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Commands: Consumer Group Subscription
//
// This example demonstrates subscribing to commands with a consumer group.
// When multiple handlers join the same group, commands are load-balanced across them
// so that only one handler in the group processes each command.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - Run Commands.SendCommand in a separate terminal to send commands
//   - dotnet run

using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Commands;
using System.Text;

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-commands-consumer-group-client",
});
await client.ConnectAsync();

Console.WriteLine("Subscribed to commands with consumer group 'handler-group'...");

var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true;
    cts.Cancel();
};

await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "csharp-commands.consumer-group", Group = "handler-group" },
    cts.Token))
{
    var body = Encoding.UTF8.GetString(cmd.Body.Span);
    Console.WriteLine($"Command: {cmd.RequestId} — {body}");

    await client.SendCommandResponseAsync(new CommandResponse
    {
        RequestId = cmd.RequestId,
        ReplyChannel = cmd.ReplyChannel!,
        Executed = true,
    });

    Console.WriteLine("  -> Responded: executed=true");
}

Console.WriteLine("Done.");

How It Works

• Group = "handler-group" in CommandsSubscription registers this subscriber as a member of a named consumer group. The broker routes each incoming command to exactly one member.
• Run multiple instances of this program (with different ClientId values) to load-balance command handling across a pool of workers.
• Each handler must call SendCommandResponseAsync with the RequestId and ReplyChannel from the incoming command — the broker uses these to route the response back to the original sender.
• The CancellationTokenSource / Console.CancelKeyPress pattern allows clean Ctrl+C shutdown without leaving commands unacknowledged in-flight.

Related

• Pattern overview
• C# SDK Reference
• Send Command
• Handle Command