Command Group
Load-balance KubeMQ commands across multiple handlers in a group using the Python SDK.
Overview
A command consumer group turns a single command handler into a scalable worker pool: run multiple identical subscribers 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 passes the same group alongside the channel on its CommandsSubscription to subscribe_to_commands; the broker tracks membership and picks one live member per command. send_command on the caller side is unaware groups exist — it just awaits a CommandResponse, which comes back from whichever handler happened to process 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 - Python SDK installed (
pip install kubemq)
Code
"""Example: Consumer group — load-balance commands across multiple handlers."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
CommandMessage,
CommandResponse,
CommandsSubscription,
)
async def main() -> None:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-commands-consumer-group-client",
) as client:
token = AsyncCancellationToken()
async def make_handler(name: str) -> None:
async for cmd in client.subscribe_to_commands(
subscription=CommandsSubscription(
channel="python-commands.consumer-group",
group="handlers",
on_receive_command_callback=lambda c: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"[{name}] Received command: {cmd.body.decode('utf-8')}")
await client.send_response(
CommandResponse(command_received=cmd, is_executed=True)
)
task1 = asyncio.create_task(make_handler("Handler-1"))
task2 = asyncio.create_task(make_handler("Handler-2"))
await asyncio.sleep(1)
for i in range(4):
response = await client.send_command(
CommandMessage(
channel="python-commands.consumer-group",
body=f"Command #{i + 1}".encode(),
timeout_in_seconds=10,
)
)
print(f"Command #{i + 1} executed: {response.is_executed}")
token.cancel()
for t in [task1, task2]:
t.cancel()
try:
await t
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
Both handlers subscribe to the same channel with group="handlers". KubeMQ delivers each command to exactly one handler in the group — this is the load-balancing behaviour. Two asyncio.Tasks run concurrently inside the same AsyncCQClient, so they share the gRPC connection. Four commands are sent sequentially; you will see them distributed between Handler-1 and Handler-2 in the output.
Related
Was this page helpful?