Work Queue

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

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

Code

"""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

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

• Pattern overview
• Python SDK Reference
• Fan-Out
• Request-Reply