Query Group
Load-balance KubeMQ queries across multiple handlers in a group using the Python SDK.
Overview
A consumer group scales query handling horizontally without touching the caller's side. Instead of one process answering every query on a channel, you run several identical handler instances under the same group name, and the broker routes each query to exactly one member — never to all of them. That turns a single responder into a pool you can grow or shrink to match load, which matters for anything RPC-shaped: a lookup service, a cache-fill handler, a synchronous read path behind an API.
It works by tying group membership to the subscription: QueriesSubscription(channel=..., group=...) passed to client.subscribe_to_queries load-balances across every subscriber sharing that channel and group. The sender calls client.send_query exactly as it would against a single handler — it never knows how many members exist or which one answered.
Gotchas: channel and group name must match exactly, or a typo quietly creates a second, empty group instead of erroring. Omit group and every subscriber reverts to broadcast, each answering independently. A stuck group member isn't bypassed — the caller just sees a timeout.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Consumer group — load-balance queries across multiple responders."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
QueriesSubscription,
QueryMessage,
QueryResponse,
)
async def main() -> None:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-queries-consumer-group-client",
) as client:
token = AsyncCancellationToken()
async def make_responder(name: str) -> None:
async for query in client.subscribe_to_queries(
subscription=QueriesSubscription(
channel="python-queries.consumer-group",
group="responders",
on_receive_query_callback=lambda q: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"[{name}] Received query: {query.body.decode('utf-8')}")
await client.send_response(
QueryResponse(
query_received=query,
is_executed=True,
body=f"Response from {name}".encode(),
)
)
task1 = asyncio.create_task(make_responder("Responder-1"))
task2 = asyncio.create_task(make_responder("Responder-2"))
await asyncio.sleep(1)
for i in range(4):
response = await client.send_query(
QueryMessage(
channel="python-queries.consumer-group",
body=f"Query #{i + 1}".encode(),
timeout_in_seconds=10,
)
)
print(f"Query #{i + 1} response: {response.body}")
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 responders subscribe to the same channel with group="responders". KubeMQ routes each query to exactly one responder in the group (round-robin or least-loaded, depending on server config). Each responder returns body=f"Response from {name}".encode() so you can see in the output which instance handled each query. The load-balancing happens entirely on the broker side — the sender always calls send_query the same way regardless of how many responders are running.
Related
Was this page helpful?