# Consumer Group (/sdks/python/how-to/events/consumer-group)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```python title="consumer_group.py"
"""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 [#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 [#related]

* [Pattern overview](/learn/events/getting-started)
* [Python SDK Reference](/sdks/python/reference/events)
* [Basic Pub/Sub](/sdks/python/tutorials/basic-pubsub)
* [Cancel Subscription](/sdks/python/how-to/events/cancel-subscription)