Consumer Group

Overview

A consumer group turns Events pub/sub from a broadcast into a work queue. By default every subscriber on a channel gets every event — fine for notifications, but wasteful when you want a pool of workers to split a stream of tasks so each one is handled exactly once. Reach for a consumer group whenever you're scaling out event processing and duplicate work isn't just wasteful but actively wrong (double-charging a customer, double-sending an alert).

It works by naming a group when you subscribe: every subscriber that passes the same group value in EventsSubscription (used with subscribe_to_events) joins that group, and the broker round-robins each event to exactly one member instead of fanning it out to all of them. Leaving group unset (or empty) reverts to normal fan-out, so the same subscription shape can flip between the two delivery models with one field.

Gotchas: consumer groups are scoped per channel — subscribing to the same group on a different channel does not share load balancing across channels. A group with zero active subscribers behaves like no subscribers at all; events aren't queued for a group that's temporarily empty the way they are for durable queue messages. And because delivery is round-robin rather than content-aware, you can't route specific events to specific workers within a group — if you need that, partition by channel instead.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Consumer group — load-balance events across 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-events-consumer-group-client",
    ) as client:
        token = AsyncCancellationToken()

        async def make_worker(name: str) -> None:
            async for event in client.subscribe_to_events(
                subscription=EventsSubscription(
                    channel="python-events.consumer-group",
                    group="workers",
                    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')}")

        task1 = asyncio.create_task(make_worker("Worker-1"))
        task2 = asyncio.create_task(make_worker("Worker-2"))
        await asyncio.sleep(1)

        for i in range(6):
            await client.publish_event(
                EventMessage(
                    channel="python-events.consumer-group",
                    body=f"Task #{i + 1}".encode(),
                )
            )

        await asyncio.sleep(3)
        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 subscribers share the same group="workers" on the same channel; KubeMQ delivers each event to exactly one subscriber in the group.
• Without a group, every subscriber receives every event (broadcast/fan-out); the group option enables competitive consumption.
• Each subscriber coroutine runs in its own asyncio task, so both are active concurrently and ready to receive.
• A single AsyncCancellationToken is shared by both tasks, so one cancel() call stops all group members simultaneously.

Related

• Pattern overview
• Python SDK Reference
• Basic Pub/Sub
• Cancel Subscription