Consumer Group

Overview

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 passes the same group string to SubscribeToEvents joins that group, and the broker round-robins each event to exactly one member instead of fanning it out to all of them. Passing an empty group string reverts to normal fan-out, so the same subscription call can flip between the two delivery models with one argument.

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
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: events/consumer-group
//
// Demonstrates load-balanced event consumption using consumer groups.
// When multiple subscribers share the same group on the same channel,
// each event is delivered to exactly one subscriber in the group.
//
// Channel: go-events.consumer-group
// Client ID: go-events-consumer-group-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

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

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-events-consumer-group-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events.consumer-group"
	group := "go-events-worker-group"
	received := make(chan struct{})

	// Subscribe with a consumer group for load-balanced delivery.
	sub, err := client.SubscribeToEvents(ctx, channel, group,
		kubemq.WithOnEvent(func(event *kubemq.Event) {
			fmt.Printf("Consumer group received: channel=%s body=%s\n",
				event.Channel, event.Body)
			close(received)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Subscription error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()

	// Publish an event to the group channel.
	err = client.SendEvent(ctx, kubemq.NewEvent().
		SetChannel(channel).
		SetBody([]byte("hello consumer group")).
		SetMetadata("group-demo"))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Event published to consumer group channel")

	select {
	case <-received:
		fmt.Println("Consumer group demo complete")
	case <-ctx.Done():
		log.Fatal("Timed out waiting for consumer group event")
	}
}

How It Works

1. SubscribeToEvents is called with group := "go-events-worker-group" — a non-empty group activates competing-consumer (load-balanced) delivery on the server.
2. With a consumer group, the broker delivers each event to exactly one subscriber within the group rather than broadcasting to all.
3. To scale throughput, run additional subscribers with the same channel and group arguments; the broker round-robins deliveries across them automatically.
4. Passing "" as the group would switch to fan-out mode, where every subscriber gets every event.

Related

• Pattern overview
• Go SDK Reference
• Basic Pub/Sub
• Cancel Subscription