Command Group
Load-balanced command handling across multiple workers
Overview
A command consumer group turns a single command handler into a scalable worker pool: run multiple identical instances subscribed with the same group name, and the broker load-balances each incoming command to exactly one member instead of broadcasting it to all of them. This is how you add capacity to handle a growing command volume — start more workers in the same group — without changing anything on the caller's side.
Every subscriber sets the same group alongside channel in its subscribeToCommands builder; the broker tracks membership and picks one live member per command. sendCommand on the caller side is unaware groups exist — it just suspends for a response, which comes back from whichever worker happened to handle it.
Gotchas: group membership is scoped per channel — subscribers on the same channel with different group names each get their own full copy of every command (fan-out), which looks like a bug when you expected load-balancing. A slow handler still holds up the caller's timeout, since only one worker is ever picked. And if every member of the group is offline when a command arrives, the send simply fails or times out — commands aren't queued or replayed for a group that has no active listener.
Prerequisites
- KubeMQ server running on
localhost:50000 - Kotlin SDK installed (
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
Code
package io.kubemq.sdk.examples.commands
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.cq.commandMessage
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-commands-command-group"
private const val CHANNEL = "kotlin-commands.command-group"
fun main() = runBlocking {
val client = KubeMQClient.cq {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createCommandsChannel(CHANNEL)
// Subscribe 3 handlers in the same consumer group (load-balanced)
val group = "command-handlers"
val numWorkers = 3
val counts = Array(numWorkers) { AtomicInteger(0) }
val jobs = (0 until numWorkers).map { i ->
launch {
client.subscribeToCommands {
channel = CHANNEL
this.group = group
}.collect { cmd ->
counts[i].incrementAndGet()
val response = cmd.respond { executed = true }
client.sendCommandResponse(response)
}
}
}
delay(500)
// Send 9 commands to the group (distributed across workers)
println("Sending 9 commands to group '$group'...\n")
repeat(9) { i ->
try {
client.sendCommand(commandMessage {
channel = CHANNEL
body = "Cmd #${i + 1}".toByteArray()
timeoutMs = 10000
})
} catch (e: Exception) {
println("Command ${i + 1} error: ${e.message}")
}
}
delay(1000)
println("Distribution:")
for (i in 0 until numWorkers) {
println(" Worker ${i + 1}: ${counts[i].get()}")
}
jobs.forEach { it.cancel() }
} finally {
try { client.deleteCommandsChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
- Setting
group = "command-handlers"enables consumer group mode for command subscriptions. - Each command is delivered to exactly one worker in the group.
- The server distributes commands across group members for load balancing.
- All workers share the same group name but handle different commands.
Related
Was this page helpful?