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

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

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

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()
}

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()

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'`);

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();

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)}");
}

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()

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));

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(())
}

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

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)

Run multiple instances with different WORKER_ID values:

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

Scaling Pattern: Group Workers + Monitor

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

# 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

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`).

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 or Queues.

Consumer Groups Tutorial for step-by-step group setup
Handle Slow Consumers for mitigation strategies
Events Reference for group and subscription configuration

Scale Subscribers Horizontally

How Channel Groups Work

Set Up a Consumer Group

Scaling Pattern: Group Workers + Monitor

Scaling Guidelines

On this page

Scale Subscribers Horizontally

How Channel Groups Work

Set Up a Consumer Group

Scaling Pattern: Group Workers + Monitor

Scaling Guidelines

Related

On this page