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 whose subscribeToEvents { } block sets the same group value joins that group, and the broker round-robins each event to exactly one member instead of fanning it out to all of them. Omitting group (or leaving it empty) reverts to normal fan-out, so the same subscription block 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
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

ConsumerGroupExample.kt

package io.kubemq.sdk.examples.events

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.eventMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.concurrent.atomic.AtomicInteger

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-events-consumer-group"
private const val CHANNEL = "kotlin-events.consumer-group"

fun main() = runBlocking {
    val client = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        val numSubscribers = 3
        val numMessages = 9
        val counts = Array(numSubscribers) { AtomicInteger(0) }
        val totalReceived = AtomicInteger(0)

        // Subscribe with consumer group -- load balancing across members
        val jobs = (0 until numSubscribers).map { i ->
            launch {
                client.subscribeToEvents {
                    channel = CHANNEL
                    group = "processors"
                }.collect { msg ->
                    counts[i].incrementAndGet()
                    println("[group-${i + 1}] ${msg.channel}: ${String(msg.body)}")
                    totalReceived.incrementAndGet()
                }
            }
        }

        delay(500)

        // Publish messages -- each delivered to exactly one subscriber in the group
        repeat(numMessages) { i ->
            client.publishEvent(eventMessage {
                channel = CHANNEL
                body = "Message #${i + 1}".toByteArray()
            })
            delay(100)
        }

        delay(2000)

        println("\nDistribution:")
        for (i in 0 until numSubscribers) {
            println("  Subscriber ${i + 1}: ${counts[i].get()} messages")
        }

        jobs.forEach { it.cancel() }
        println("Done.")
    }
}

How It Works

Setting group = "processors" in the subscription config enables consumer group mode.
Each message is delivered to exactly one subscriber in the group (load balancing).
Without a group, messages are broadcast to all subscribers (fan-out).
The server distributes messages across group members automatically.

Consumer Group

Overview

Prerequisites

Code

How It Works

On this page

Consumer Group

Overview

Prerequisites

Code

How It Works

Related

On this page