# Scale Subscribers Horizontally (/learn/events/how-to/scale-subscribers)



When a single subscriber cannot keep up with event throughput, KubeMQ **channel groups** let you distribute events across multiple instances. Within a group, each event is delivered to exactly one member (round-robin).

## How Channel Groups Work [#how-channel-groups-work]

<Mermaid
  chart="flowchart LR
    P[&#x22;Publisher&#x22;]
    CH{{&#x22;Events channel<br/>order-events&#x22;}}
    G1A[&#x22;Worker A<br/>group: processors&#x22;]
    G1B[&#x22;Worker B<br/>group: processors&#x22;]
    G1C[&#x22;Worker C<br/>group: processors&#x22;]
    M[&#x22;Monitor<br/>no group&#x22;]

    P -->|publish| CH
    CH -->|round-robin| G1A
    CH -->|round-robin| G1B
    CH -->|round-robin| G1C
    CH -->|all events| M

    class CH events
    class P,G1A,G1B,G1C,M client"
/>

*Grouped workers in `processors` split the load round-robin; the ungrouped monitor still receives every event.*

* **Grouped subscribers** share the load: each event goes to exactly one member
* **Ungrouped subscribers** receive every event (standard fan-out)
* Groups are independent per channel

## Set Up a Consumer Group [#set-up-a-consumer-group]

The `group` parameter determines group membership. Subscribers with the same group name on the same channel form a group.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="grouped_worker.go"
    package main

    import (
        "context"
        "fmt"
        "log"
        "os"

        "github.com/kubemq-io/kubemq-go/v2"
    )

    func main() {
        workerID := os.Getenv("WORKER_ID")
        if workerID == "" {
            workerID = "worker-1"
        }

        ctx := context.Background()
        client, err := kubemq.NewClient(ctx,
            kubemq.WithAddress("localhost", 50000),
        )
        if err != nil {
            log.Fatal(err)
        }
        defer client.Close()

        sub, err := client.SubscribeToEvents(ctx, "order-events", "processors",
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[%s] Processing: %s\n", workerID,
                    string(event.Body))
            }),
            kubemq.WithOnError(func(err error) {
                log.Printf("[%s] Error: %v", workerID, err)
            }),
        )
        if err != nil {
            log.Fatal(err)
        }
        defer sub.Unsubscribe()

        log.Printf("[%s] Ready in group 'processors'", workerID)
        <-ctx.Done()
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="grouped_worker.py"
    import os
    import time
    from kubemq.pubsub import Client as PubSubClient
    from kubemq.pubsub import EventsSubscription, CancellationToken

    worker_id = os.environ.get("WORKER_ID", "worker-1")

    def on_event(event):
        print(f"[{worker_id}] Processing: {event.body.decode('utf-8')}")

    client = PubSubClient(address="localhost:50000")
    client.subscribe_to_events(
        subscription=EventsSubscription(
            channel="order-events",
            group="processors",
            on_receive_event_callback=on_event,
            on_error_callback=lambda e: print(f"[{worker_id}] Error: {e}"),
        ),
        cancel=CancellationToken(),
    )
    print(f"[{worker_id}] Ready in group 'processors'")
    time.sleep(300)
    client.close()
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="grouped_worker.js"
    const { KubeMQClient } = require("kubemq-js");

    const workerId = process.env.WORKER_ID ?? "worker-1";
    const client = new KubeMQClient({ address: "localhost:50000" });

    client.subscribeToEvents({
      channel: "order-events",
      group: "processors",
      onEvent: (msg) =>
        console.log(
          `[${workerId}] Processing: ${Buffer.from(msg.body).toString()}`
        ),
      onError: (err) =>
        console.error(`[${workerId}] Error:`, err.message),
    });

    console.log(`[${workerId}] Ready in group 'processors'`);
    ```
  </Tab>

  <Tab value="Java">
    ```java title="GroupedWorker.java"
    String workerId = System.getenv().getOrDefault("WORKER_ID", "worker-1");

    PubSubClient client = PubSubClient.builder()
        .address("localhost:50000")
        .clientId(workerId)
        .build();

    client.subscribeToEvents(EventsSubscription.builder()
        .channel("order-events")
        .group("processors")
        .onReceiveEventCallback(event ->
            System.out.printf("[%s] Processing: %s%n", workerId,
                new String(event.getBody())))
        .onErrorCallback(err ->
            System.err.printf("[%s] Error: %s%n", workerId, err.getMessage()))
        .build());

    System.out.printf("[%s] Ready in group 'processors'%n", workerId);
    Thread.sleep(300_000);
    client.close();
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="GroupedWorker.cs"
    var workerId = Environment.GetEnvironmentVariable("WORKER_ID") ?? "worker-1";

    await using var client = new KubeMQClient(new KubeMQClientOptions());
    await client.ConnectAsync();

    Console.WriteLine($"[{workerId}] Ready in group 'processors'");
    await foreach (var msg in client.SubscribeToEventsAsync(
        new EventsSubscription { Channel = "order-events", Group = "processors" }))
    {
        Console.WriteLine($"[{workerId}] Processing: "
            + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="GroupedWorker.kt"
    val workerId = System.getenv("WORKER_ID") ?: "worker-1"
    val client = PubSubClient("localhost:50000")

    client.subscribeToEvents(
        channel = "order-events",
        group = "processors",
        onEvent = { event ->
            println("[$workerId] Processing: ${String(event.body)}")
        },
        onError = { err ->
            System.err.println("[$workerId] Error: ${err.message}")
        }
    )

    println("[$workerId] Ready in group 'processors'")
    Thread.sleep(300_000)
    client.close()
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="grouped_worker.cpp"
    auto workerId = std::getenv("WORKER_ID") ?
        std::string(std::getenv("WORKER_ID")) : std::string("worker-1");

    auto client = kubemq::PubSubClient("localhost:50000");

    client.subscribeToEvents("order-events", "processors",
        [&workerId](const kubemq::Event& event) {
            std::cout << "[" << workerId << "] Processing: "
                      << event.body << std::endl;
        },
        [&workerId](const std::string& err) {
            std::cerr << "[" << workerId << "] Error: " << err << std::endl;
        }
    );

    std::cout << "[" << workerId << "] Ready in group 'processors'" << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(300));
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="grouped_worker.rs"
    use kubemq::prelude::*;
    use std::time::Duration;

    #[tokio::main]
    async fn main() -> kubemq::Result<()> {
        let worker_id = std::env::var("WORKER_ID")
            .unwrap_or_else(|_| "worker-1".to_string());

        let client = KubemqClient::builder()
            .host("localhost")
            .port(50000)
            .build()
            .await?;

        // Same group "processors" on the same channel → each event goes to one member
        let sub = client
            .subscribe_to_events(
                "order-events",
                "processors",
                move |event| {
                    let worker_id = worker_id.clone();
                    Box::pin(async move {
                        println!(
                            "[{}] Processing: {}",
                            worker_id,
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;

        tokio::time::sleep(Duration::from_secs(300)).await;

        sub.unsubscribe().await;
        client.close().await?;
        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="grouped_worker.rb"
    require 'kubemq'

    worker_id = ENV.fetch('WORKER_ID', 'worker-1')

    client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: worker_id)
    cancel = KubeMQ::CancellationToken.new

    subscription = KubeMQ::PubSub::EventsSubscription.new(
      channel: 'order-events',
      group: 'processors'
    )
    client.subscribe_to_events(subscription, cancellation_token: cancel,
      on_error: lambda { |e| warn "[#{worker_id}] Error: #{e.message}" }) do |event|
      puts "[#{worker_id}] Processing: #{event.body}"
    end

    puts "[#{worker_id}] Ready in group 'processors'"
    sleep 300
    cancel.cancel
    client.close
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="grouped_worker.exs"
    worker_id = System.get_env("WORKER_ID", "worker-1")

    {:ok, client} =
      KubeMQ.Client.start_link(address: "localhost:50000", client_id: worker_id)

    # Same group "processors" on the same channel → each event goes to one member
    {:ok, sub} =
      KubeMQ.Client.subscribe_to_events(client, "order-events",
        group: "processors",
        on_event: fn event ->
          IO.puts("[#{worker_id}] Processing: #{event.body}")
        end,
        on_error: fn err ->
          IO.puts("[#{worker_id}] Error: #{inspect(err)}")
        end
      )

    IO.puts("[#{worker_id}] Ready in group 'processors'")
    Process.sleep(300_000)

    KubeMQ.Subscription.cancel(sub)
    KubeMQ.Client.close(client)
    ```
  </Tab>
</Tabs>

Run multiple instances with different `WORKER_ID` values:

```bash
WORKER_ID=worker-A ./grouped_worker &
WORKER_ID=worker-B ./grouped_worker &
WORKER_ID=worker-C ./grouped_worker &
```

## Scaling Pattern: Group Workers + Monitor [#scaling-pattern-group-workers--monitor]

Combine grouped workers with an ungrouped monitor that sees all events:

```bash
# Workers (group: processors) — each gets ~1/3 of events
WORKER_ID=worker-A ./grouped_worker &
WORKER_ID=worker-B ./grouped_worker &
WORKER_ID=worker-C ./grouped_worker &

# Monitor (no group) — receives ALL events
./monitor &
```

## Scaling Guidelines [#scaling-guidelines]

| Factor                  | Recommendation                                                                  |
| ----------------------- | ------------------------------------------------------------------------------- |
| Number of group members | Scale horizontally based on throughput. No hard limit.                          |
| Multiple groups         | Different groups on the same channel each get full delivery.                    |
| Slow consumers          | Events are dropped after the write deadline (2s default). Keep processing fast. |
| Group naming            | Use descriptive names (e.g., `email-senders`, `analytics-workers`).             |

<Callout type="warn">
  Channel groups provide **load balancing**, not guaranteed delivery. If a grouped subscriber disconnects, events routed to it are lost. For guaranteed delivery, use [Events Store consumer groups](/learn/events-store) or [Queues](/learn/queues).
</Callout>

## Related [#related]

* [Consumer Groups Tutorial](/learn/events/tutorials/consumer-groups) for step-by-step group setup
* [Handle Slow Consumers](/learn/events/how-to/handle-slow-consumers) for mitigation strategies
* [Events Reference](/learn/events/reference) for group and subscription configuration