# Load Balance Across Responders (/learn/rpc/how-to/load-balancing)



## How Load Balancing Works [#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.

<Mermaid
  chart="graph LR
    S[&#x22;Sender&#x22;]
    K[&#x22;KubeMQ&#x22;]
    R1[&#x22;Responder 1&#x22;]
    R2[&#x22;Responder 2&#x22;]
    R3[&#x22;Responder 3&#x22;]

    S -- &#x22;send command&#x22; --> K
    K -- &#x22;round-robin&#x22; --> R1
    K -- &#x22;round-robin&#x22; --> R2
    K -- &#x22;round-robin&#x22; --> R3

    subgraph group[&#x22;Group: order-workers&#x22;]
        R1
        R2
        R3
    end

    class S client
    class K broker
    class R1,R2,R3 command"
/>

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

## Steps [#steps]

<Steps>
  <Step>
    ### Deploy Multiple Responders with Same Group [#deploy-multiple-responders-with-same-group]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
            }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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,
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="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),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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
            });
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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}") }
        )
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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;
            }
        );
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
          )
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Send Requests [#send-requests]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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}")
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="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}`);
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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());
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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;
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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);
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
        ```
      </Tab>
    </Tabs>
  </Step>
</Steps>

## Group Behavior [#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               |

<Callout type="info">
  Groups work the same way for both commands and queries. Use the `group` parameter when subscribing.
</Callout>

## Scaling Strategy [#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