Overview

This covers the load-balance delivery mode

For an overview of both delivery models, see Multiple Subscribers. For the broadcast mode instead, see Fan-Out.

A consumer group turns Events pub/sub from a broadcast into a work queue. By default every subscriber on a channel gets every event — fine for notifications, but wasteful when you want a pool of workers to split a stream of tasks so each one is handled exactly once. Reach for a consumer group whenever you're scaling out event processing and duplicate work isn't just wasteful but actively wrong (double-charging a customer, double-sending an alert).

It works by naming a group when you subscribe: every subscriber that sets the same Group value on EventsSubscription (passed to SubscribeToEventsAsync) joins that group, and the broker routes each event to exactly one member instead of fanning it out to all of them. Leaving Group unset reverts to normal broadcast semantics, so the same subscription shape can flip between the two delivery models with one property.

Gotchas: consumer groups are scoped per channel — subscribing to the same group on a different channel does not share load balancing across channels. A group with zero active subscribers behaves like no subscribers at all; events aren't queued for a group that's temporarily empty the way they are for durable queue messages. And because delivery is round-robin rather than content-aware, you can't route specific events to specific workers within a group — if you need that, partition by channel instead.

Prerequisites

KubeMQ server running on localhost:50000
C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

Program.cs

// KubeMQ .NET SDK — Events: Consumer Group Subscription
//
// This example demonstrates subscribing to events with a consumer group.
// When multiple subscribers join the same group, events are load-balanced
// across them so that only one subscriber in the group receives each event.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

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

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

Console.WriteLine("Connected to KubeMQ server");

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

var subscribeTask = Task.Run(async () =>
{
    await foreach (var msg in client.SubscribeToEventsAsync(
        new EventsSubscription { Channel = "csharp-events.consumer-group", Group = "my-group" },
        cts.Token))
    {
        var body = Encoding.UTF8.GetString(msg.Body.Span);
        Console.WriteLine($"Received event: {body}");
    }
});

await Task.Delay(1000);

for (var i = 1; i <= 5; i++)
{
    await client.SendEventAsync(new EventMessage
    {
        Channel = "csharp-events.consumer-group",
        Body = Encoding.UTF8.GetBytes($"Group Event #{i}"),
    });
    Console.WriteLine($"Published event #{i}");
}

await Task.Delay(2000);
cts.Cancel();

Console.WriteLine("Done.");

How It Works

Setting Group = "my-group" on EventsSubscription registers this subscriber as a member of a named consumer group. The broker routes each incoming event to exactly one member of the group.
Run a second instance of this program (with a different ClientId) to observe load-balancing: each event goes to only one subscriber.
Without Group, every subscriber receives every event (broadcast semantics). With a group, you get work-queue semantics.
The CancellationTokenSource / Console.CancelKeyPress pattern wires Ctrl+C to a cooperative shutdown of the await foreach loop.

Consumer Group

Overview

Prerequisites

Code

How It Works

Related

On this page