Multiple Subscribers
Deliver the same KubeMQ event to multiple independent subscribers at once using the Python SDK.
Overview
Fan-out delivery lets several independent consumers each get their own copy of every event published on a channel — the pattern behind broadcasting a notification to every connected service or feeding the same stream to a cache invalidator and a metrics collector at once. Reach for it whenever multiple, unrelated pieces of code all need to react to the same event, rather than compete for it.
It works by calling subscribe_to_events more than once for the same channel while leaving group empty (""). Each call opens its own stream, and the broker treats every subscriber with no group as broadcast: publishing one event delivers it to every open stream — the opposite of a consumer group, where subscribers sharing a group name split events among themselves for load balancing.
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. Mixing a non-empty group into one subscriber on the same channel silently turns broadcast into load-balancing for it. And because delivery is fully concurrent, shared state your callbacks touch needs its own synchronization.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Multiple subscribers — demonstrates broadcast and group load balancing."""
from __future__ import annotations
import asyncio
from kubemq import AsyncCancellationToken, AsyncPubSubClient, EventMessage, EventsSubscription
async def main() -> None:
async with AsyncPubSubClient(
address="localhost:50000",
client_id="python-events-multiple-subscribers-client",
) as client:
token = AsyncCancellationToken()
tasks: list[asyncio.Task[None]] = []
async def make_subscriber(channel: str, name: str, group: str = "") -> None:
async for event in client.subscribe_to_events(
subscription=EventsSubscription(
channel=channel,
group=group,
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"[{name}] Received: {event.body.decode('utf-8')}")
# Broadcast: all subscribers receive every message
tasks.append(asyncio.create_task(
make_subscriber("python-events.multiple-subscribers", "Subscriber-A")
))
tasks.append(asyncio.create_task(
make_subscriber("python-events.multiple-subscribers", "Subscriber-B")
))
# Group subscription: only one subscriber in the group receives each message
tasks.append(asyncio.create_task(
make_subscriber("python-events.multiple-subscribers-tasks", "Worker-1", "workers")
))
tasks.append(asyncio.create_task(
make_subscriber("python-events.multiple-subscribers-tasks", "Worker-2", "workers")
))
await asyncio.sleep(1)
# Broadcast: both Subscriber-A and Subscriber-B receive this
await client.publish_event(
EventMessage(
channel="python-events.multiple-subscribers",
body=b"System update available",
)
)
# Group: only one of Worker-1 or Worker-2 receives each task
for i in range(4):
await client.publish_event(
EventMessage(
channel="python-events.multiple-subscribers-tasks",
body=f"Task #{i + 1}".encode(),
)
)
await asyncio.sleep(3)
token.cancel()
for t in tasks:
t.cancel()
try:
await t
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
- Subscriber-A and Subscriber-B subscribe to the same channel with no
group, so both receive every published event (broadcast). - Worker-1 and Worker-2 subscribe to a separate channel with
group="workers", so only one receives each event (load-balanced). - All four subscriber coroutines run concurrently via
asyncio.create_task, sharing the singleAsyncPubSubClientconnection. - A shared
AsyncCancellationTokenstops all subscriptions at once; each task is also individually cancelled to unblock pending awaits.
Related
Was this page helpful?