How Load Balancing Works

When multiple responders subscribe to the same channel with the same group name, KubeMQ distributes requests across them in round-robin fashion. Each request goes to exactly one responder in the group.

Responders sharing a group act as competing consumers — KubeMQ routes each request to exactly one member.

Steps

Deploy Multiple Responders with Same Group

Each responder subscribes with group="order-workers". KubeMQ routes each request to one member of the group.

responder.go

workerId := os.Getenv("WORKER_ID")

_, err := client.SubscribeToCommands(ctx, "orders.process", "order-workers",
    kubemq.WithOnCommandReceive(func(cmd *kubemq.CommandReceive) {
        fmt.Printf("[Worker %s] Processing: %s\n", workerId, cmd.Body)
        resp := kubemq.NewCommandReply().
            SetRequestId(cmd.Id).
            SetResponseTo(cmd.ResponseTo).
            SetExecutedAt(time.Now())
        _ = client.SendCommandResponse(ctx, resp)
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("Error:", err)
    }),
)

responder.py

import os

worker_id = os.getenv("WORKER_ID", "1")

def on_command(request):
    print(f"[Worker {worker_id}] Processing: {request.body.decode('utf-8')}")
    client.send_response_message(
        CommandResponse(command_received=request, is_executed=True)
    )

client.subscribe_to_commands(
    subscription=CommandsSubscription(
        channel="orders.process",
        group="order-workers",
        on_receive_command_callback=on_command,
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=cancel,
)

responder.js

const workerId = process.env.WORKER_ID || "1";

client.subscribeToCommands({
  channel: "orders.process",
  group: "order-workers",
  onCommand: (cmd) => {
    console.log(`[Worker ${workerId}] Processing:`, Buffer.from(cmd.body).toString());
    client.sendCommandResponse({ id: cmd.id, replyChannel: cmd.replyChannel, executed: true });
  },
  onError: (err) => console.error("Error:", err.message),
});

Responder.java

String workerId = System.getenv("WORKER_ID");

client.subscribeToCommands(CommandsSubscription.builder()
    .channel("orders.process")
    .group("order-workers")
    .onReceiveCommandCallback(cmd -> {
        System.out.printf("[Worker %s] Processing: %s%n",
            workerId, new String(cmd.getBody()));
        return CommandResponseMessage.builder()
            .requestId(cmd.getId()).isExecuted(true).build();
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

Responder.cs

var workerId = Environment.GetEnvironmentVariable("WORKER_ID") ?? "1";

await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "orders.process", Group = "order-workers" }))
{
    Console.WriteLine($"[Worker {workerId}] Processing: "
        + Encoding.UTF8.GetString(cmd.Body.Span));
    await client.SendCommandResponseAsync(new CommandResponse
    {
        RequestId = cmd.Id, IsExecuted = true
    });
}

Responder.kt

val workerId = System.getenv("WORKER_ID") ?: "1"

client.subscribeToCommands(
    channel = "orders.process",
    group = "order-workers",
    onCommand = { cmd ->
        println("[Worker $workerId] Processing: ${String(cmd.body)}")
        client.sendCommandResponse(requestId = cmd.id, isExecuted = true)
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

responder.cpp

std::string workerId = std::getenv("WORKER_ID") ? std::getenv("WORKER_ID") : "1";

client.subscribeToCommands("orders.process", "order-workers",
    [&](const kubemq::CommandReceive& cmd) {
        std::cout << "[Worker " << workerId << "] Processing: "
                  << cmd.body << std::endl;
        client.sendCommandResponse(cmd.id, true);
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

responder.rs

let worker_id = std::env::var("WORKER_ID").unwrap_or_else(|_| "1".into());

let rc = client.clone();
let sub = client
    .subscribe_to_commands(
        "orders.process",
        "order-workers", // shared group → competing consumers
        move |cmd| {
            let c = rc.clone();
            let id = worker_id.clone();
            Box::pin(async move {
                println!("[Worker {}] Processing: {}", id, String::from_utf8_lossy(&cmd.body));
                let reply = CommandReplyBuilder::new()
                    .request_id(&cmd.id)
                    .response_to(&cmd.response_to)
                    .build();
                let _ = c.send_command_response(reply).await;
            })
        },
        None,
    )
    .await?;

responder.rb

worker_id = ENV.fetch("WORKER_ID", "1")

sub = KubeMQ::CQ::CommandsSubscription.new(channel: "orders.process", group: "order-workers")
client.subscribe_to_commands(sub, cancellation_token: cancel,
                             on_error: ->(e) { puts "Error: #{e.message}" }) do |cmd|
  puts "[Worker #{worker_id}] Processing: #{cmd.body}"
  client.send_response(KubeMQ::CQ::CommandResponseMessage.new(
    request_id: cmd.id, reply_channel: cmd.reply_channel, executed: true
  ))
end

responder.exs

worker_id = System.get_env("WORKER_ID", "1")

{:ok, sub} =
  KubeMQ.Client.subscribe_to_commands(client, "orders.process",
    group: "order-workers",
    on_command: fn cmd ->
      IO.puts("[Worker #{worker_id}] Processing: #{cmd.body}")
      KubeMQ.CommandReply.new(
        request_id: cmd.id, response_to: cmd.reply_channel, executed: true)
    end
  )

Send Requests

The sender does not need any changes — KubeMQ handles the distribution automatically.

sender.go

for i := 0; i < 10; i++ {
    resp, err := client.SendCommand(ctx, kubemq.NewCommand().
        SetChannel("orders.process").
        SetBody([]byte(fmt.Sprintf(`{"orderId":"ORD-%04d"}`, i))).
        SetTimeout(10 * time.Second))
    if err != nil {
        log.Printf("Request %d failed: %v", i, err)
        continue
    }
    log.Printf("Request %d — Executed: %v", i, resp.Executed)
}

sender.py

for i in range(10):
    response = client.send_command(
        CommandMessage(
            channel="orders.process",
            body=f'{{"orderId":"ORD-{i:04d}"}}'.encode(),
            timeout_in_seconds=10,
        )
    )
    print(f"Request {i} — Executed: {response.is_executed}")

sender.js

for (let i = 0; i < 10; i++) {
  const response = await client.sendCommand({
    channel: "orders.process",
    body: Buffer.from(JSON.stringify({ orderId: `ORD-${String(i).padStart(4, "0")}` })),
    timeoutInSeconds: 10,
  });
  console.log(`Request ${i} — Executed: ${response.executed}`);
}

Sender.java

for (int i = 0; i < 10; i++) {
    var resp = client.sendCommandRequest(CommandMessage.builder()
        .channel("orders.process")
        .body(String.format("{\"orderId\":\"ORD-%04d\"}", i).getBytes())
        .timeout(10000).build());
    System.out.printf("Request %d — Executed: %s%n", i, resp.isExecuted());
}

Sender.cs

for (int i = 0; i < 10; i++)
{
    var resp = await client.SendCommandAsync(new CommandMessage
    {
        Channel = "orders.process",
        Body = Encoding.UTF8.GetBytes($"{{\"orderId\":\"ORD-{i:D4}\"}}"),
        Timeout = TimeSpan.FromSeconds(10)
    });
    Console.WriteLine($"Request {i} — Executed: {resp.IsExecuted}");
}

Sender.kt

repeat(10) { i ->
    val resp = client.sendCommand(CommandMessage(
        channel = "orders.process",
        body = """{"orderId":"ORD-${"%04d".format(i)}"}""".toByteArray(),
        timeout = 10000))
    println("Request $i — Executed: ${resp.isExecuted}")
}

sender.cpp

for (int i = 0; i < 10; i++) {
    kubemq::CommandMessage cmd;
    cmd.channel = "orders.process";
    cmd.body = R"({"orderId":"ORD-)" + std::to_string(i) + R"("})";
    cmd.timeout = 10000;

    auto resp = client.sendCommand(cmd);
    std::cout << "Request " << i << " — Executed: " << resp.isExecuted << std::endl;
}

sender.rs

for i in 0..10 {
    let command = CommandBuilder::new()
        .channel("orders.process")
        .body(format!(r#"{{"orderId":"ORD-{:04}"}}"#, i).into_bytes())
        .timeout(Duration::from_secs(10))
        .build();
    let resp = client.send_command(command).await?;
    println!("Request {} — Executed: {}", i, resp.executed);
}

sender.rb

10.times do |i|
  msg = KubeMQ::CQ::CommandMessage.new(
    channel: "orders.process",
    timeout: 10,
    body: format('{"orderId":"ORD-%04d"}', i)
  )
  result = client.send_command(msg)
  puts "Request #{i} — Executed: #{result.executed}"
end

sender.exs

for i <- 0..9 do
  cmd = KubeMQ.Command.new(
    channel: "orders.process",
    body: ~s({"orderId":"ORD-#{String.pad_leading(to_string(i), 4, "0")}"}),
    timeout: 10_000
  )

  case KubeMQ.Client.send_command(client, cmd) do
    {:ok, resp} -> IO.puts("Request #{i} — Executed: #{resp.executed}")
    {:error, err} -> IO.puts("Request #{i} failed: #{err.message}")
  end
end

Group Behavior

Configuration	Behavior
Same group name	Competing consumers — each request goes to exactly one responder (round-robin)
Empty group (`""`)	Fan-out — every responder receives every request
Different group names	Independent pools — each group gets its own copy of each request

Groups work the same way for both commands and queries. Use the group parameter when subscribing.

Scaling Strategy

Add responders to the same group to increase throughput
All responders must be stateless — any responder should handle any request
Responders can be added or removed dynamically without affecting the sender
KubeMQ automatically rebalances when group membership changes

Load Balance Across Responders