# Work Queue (/sdks/python/how-to/work-queue)



## Overview [#overview]

A **work queue** distributes a stream of tasks across a pool of workers so each task is handled exactly once, instead of every worker doing every task — the pattern you reach for whenever you need to parallelize processing (image resizing, batch jobs, background work) without coordinating which worker owns which item. The queue itself does that coordination: workers just keep polling, and the broker load-balances whatever is next in line across whichever workers happen to be asking.

`receive_queue_messages` pulls a batch bounded by `max_messages` and blocks up to `wait_timeout_seconds` if the queue is empty, so a worker long-polls instead of busy-looping or hanging forever. Delivery is competing-consumer: once one worker's call returns a message, no other worker gets it. Whether a message is removed immediately or held until confirmed determines the delivery guarantee — calling `msg.async_ack()` after processing holds the message invisible until acknowledged and redelivers it after the visibility window if a worker crashes first (at-least-once); skipping it (or auto-acking) marks the message done the instant it's handed over (at-most-once).

**Gotchas:** a worker that pulls a full `max_messages` batch and then crashes before acking every item in it leaves the unacked ones to be redelivered — possibly to a different worker — so size batches to what you can safely redo. A short `wait_timeout_seconds` turns polling into a busy-loop that hammers the broker for empty results; too long delays workers noticing new work. And skipping the explicit ack trades safety for simplicity — fine for idempotent, low-value tasks, wrong for anything that must survive a worker crash mid-task.

## Prerequisites [#prerequisites]

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

## Code [#code]

```python title="work_queue.py"
"""Example: Work queue pattern — distribute tasks across workers using queues."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def worker(worker_id: int) -> None:
    """Worker that pulls and processes tasks from the queue."""
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id=f"python-patterns-work-queue-worker-{worker_id}",
    ) as client:
        response = await client.receive_queue_messages(
            channel="python-patterns.work-queue",
            max_messages=5,
            wait_timeout_seconds=10,
        )
        for msg in response.messages:
            body = msg.body.decode("utf-8")
            print(f"  [Worker-{worker_id}] Processing: {body}")
            await asyncio.sleep(0.1)  # Simulate work
            await msg.async_ack()
            print(f"  [Worker-{worker_id}] Done: {body}")


async def main() -> None:
    # Producer: enqueue tasks
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-patterns-work-queue-producer",
    ) as client:
        for i in range(10):
            await client.send_queue_message(
                QueueMessage(
                    channel="python-patterns.work-queue",
                    body=f"Task #{i + 1}".encode(),
                )
            )
        print("Enqueued 10 tasks")

    # Start multiple workers to process tasks concurrently
    tasks = [asyncio.create_task(worker(i)) for i in range(3)]
    await asyncio.gather(*tasks)
    print("All workers finished")


if __name__ == "__main__":
    asyncio.run(main())

```

## How It Works [#how-it-works]

The producer uses `send_queue_message` in a loop to enqueue 10 tasks, then closes its client. Each `worker()` coroutine opens its own `AsyncQueuesClient`, calls `receive_queue_messages` with `max_messages=5` and `wait_timeout_seconds=10`, then processes and `async_ack()`s each message. Three workers run concurrently via `asyncio.gather`; KubeMQ delivers each message to exactly one worker (competing-consumer semantics), so the 10 tasks are distributed across the three workers.

## Related [#related]

* [Pattern overview](/learn/guides/choosing-a-pattern)
* [Python SDK Reference](/sdks/python/reference)
* [Fan-Out](/sdks/python/how-to/fan-out)
* [Request-Reply](/sdks/python/how-to/request-reply)