Fan-Out
Fan out a single KubeMQ event to multiple subscribers at once using the Python SDK pub/sub API.
Overview
Fan-out is the default delivery behavior of KubeMQ Events pub/sub: when subscribers don't join a consumer group, every subscriber gets its own independent copy of each published event. Reach for it whenever several unrelated services need to react to the same occurrence — an order placed, a config change, an audit event — without the publisher knowing or caring who's listening, and without one subscriber's slowness affecting another's delivery.
The mechanism is simply omission: calling subscribe_to_events with an EventsSubscription that has no group puts that subscription in broadcast mode instead of load-balanced mode. publish_event doesn't change at all — the publisher sends once, and the broker independently pushes a copy to every active subscriber on the channel.
Gotchas: fan-out is opt-out by default, so a typo'd or accidentally shared group value silently turns broadcast into competing-consumer load-balancing with no error raised. Events are not persisted — a subscriber whose async generator hasn't started iterating yet when publish_event runs misses that event permanently (use Events Store if you need replay). And publish_event returns as soon as the broker accepts it, not after subscribers process it, so a publisher can outrun subscription setup on a cold start — hence the short asyncio.sleep before publishing in this sample.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Fan-out pattern — broadcast a message to multiple subscribers."""
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-patterns-fan-out-client",
) as client:
token = AsyncCancellationToken()
received_counts: dict[str, int] = {"Service-A": 0, "Service-B": 0, "Service-C": 0}
# Create 3 independent subscribers (no group — each gets every message)
async def make_subscriber(name: str) -> None:
async for event in client.subscribe_to_events(
subscription=EventsSubscription(
channel="python-patterns.fan-out",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
received_counts[name] += 1
print(f" [{name}] Received: {event.body.decode('utf-8')}")
tasks = [
asyncio.create_task(make_subscriber("Service-A")),
asyncio.create_task(make_subscriber("Service-B")),
asyncio.create_task(make_subscriber("Service-C")),
]
await asyncio.sleep(1)
# Publish a single message — all 3 subscribers should receive it
print("Publishing message to fan-out channel...")
await client.publish_event(
EventMessage(
channel="python-patterns.fan-out",
body=b"Order #1001 placed",
)
)
await asyncio.sleep(3)
print(
f"\nEach subscriber received: A={received_counts['Service-A']}, "
f"B={received_counts['Service-B']}, C={received_counts['Service-C']}"
)
token.cancel()
for t in tasks:
t.cancel()
try:
await t
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
Three subscribe_to_events async generators run concurrently as separate asyncio.Tasks on the same AsyncPubSubClient, all subscribing to the same channel without a group. Because no consumer group is set, KubeMQ delivers each published message to every subscriber independently — that is the fan-out. When publish_event is called once with body=b"Order #1001 placed", all three tasks receive it and increment their respective counter in received_counts. The shared AsyncCancellationToken stops all three generators at once.
Related
Was this page helpful?