Multiple Subscribers
Broadcast events to multiple independent subscribers
Overview
Two delivery models, one page
This page is the overview of subscribing more than one consumer to the same channel. KubeMQ Events give you two distinct delivery models — pick the one your scenario needs:
- Fan-Out — broadcast: every subscriber gets its own copy of each event.
- Consumer Group — load-balance: subscribers sharing a group name split events among themselves.
When multiple consumers subscribe to the same channel, which delivery model you get depends on the group property passed to subscribeToEvents. Leaving group unset opens an independent flow per call — broadcast, where every subscriber sees every event. Setting the same group value on multiple subscribers pools them into one logical worker — load balancing, where the broker routes each event to only one member. The example below shows two subscribers with group unset, so you can see the broadcast side in action; see the linked pages above for the full write-up of each mode.
Gotchas: Events pub/sub has no durability — a subscriber that hasn't finished subscribing yet, or that disconnects, simply misses events published in that window; there's no redelivery. Setting group on one subscriber on the same channel silently turns broadcast into load-balancing for it. And because each subscription collects on its own coroutine, shared state your callbacks touch needs its own synchronization.
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.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-multiple-subscribers"
private const val CHANNEL = "kotlin-events.multiple-subscribers"
fun main() = runBlocking {
val client = KubeMQClient.pubSub {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
val numMessages = 3
val sub1Count = AtomicInteger(0)
val sub2Count = AtomicInteger(0)
val totalReceived = AtomicInteger(0)
// Two independent subscribers (no group = broadcast to all)
val job1 = launch {
client.subscribeToEvents {
channel = CHANNEL
}.collect { msg ->
sub1Count.incrementAndGet()
println(" Subscriber A: ${String(msg.body)}")
totalReceived.incrementAndGet()
}
}
val job2 = launch {
client.subscribeToEvents {
channel = CHANNEL
}.collect { msg ->
sub2Count.incrementAndGet()
println(" Subscriber B: ${String(msg.body)}")
totalReceived.incrementAndGet()
}
}
println("Two independent subscribers created (broadcast mode).\n")
delay(500)
// Send event messages (each subscriber receives all)
repeat(numMessages) { i ->
client.publishEvent(eventMessage {
channel = CHANNEL
body = "Broadcast message #${i + 1}".toByteArray()
})
delay(100)
}
delay(2000)
println("\nResults:")
println(" Subscriber A received: ${sub1Count.get()}")
println(" Subscriber B received: ${sub2Count.get()}")
println(" Both received all $numMessages messages (broadcast).")
job1.cancel()
job2.cancel()
println("Done.")
}
}How It Works
- Without a
groupproperty, each subscriber receives all published messages (broadcast/fan-out). - Both subscribers A and B receive the same 3 messages independently.
- This contrasts with consumer groups where each message goes to exactly one member.
- Use broadcast mode for notifications, audit logs, or any scenario where all consumers need every message.
Related
Was this page helpful?