Pattern 2: Producer/Consumer Queue

The Problem

Scenario: Webhooks arrive at 1000 events/sec. You have 3 workers that process events at 100 events/sec each (300 total/sec).

Without a queue: Build up a giant list? Lose events? Block the webhook handler?

With a queue: Webhooks enqueue instantly. Workers dequeue and process. Queue acts as a buffer.

Real-World Triggers

Webhook events (Stripe, GitHub, Slack)
Message queues (RabbitMQ, Kafka)
File upload processing
Background job processing (Celery, Bull)
Image resizing/thumbnail generation

How It Works: Producer → Queue → Workers

flowchart LR P["📥 Producer
(webhook handler)"] -->|"put()"| Q subgraph Q["Queue (maxsize=5)"] direction TB J1["Job 1"] J2["Job 2"] J3["Job 3 ..."] end Q -->|"get()"| W0["⚙️ Worker 0
processing"] Q -->|"get()"| W1["⚙️ Worker 1
processing"] Q -->|"get()"| W2["⚙️ Worker 2
processing"] style P fill:#fef3c7,stroke:#d97706 style Q fill:#e0e7ff,stroke:#4f46e5 style W0 fill:#dcfce7,stroke:#16a34a style W1 fill:#dcfce7,stroke:#16a34a style W2 fill:#dcfce7,stroke:#16a34a

Timeline: 100 jobs, producer at 1000/sec, 3 workers at ~9 jobs/sec total

timeline title Producer outpaces workers → queue buffers the burst section t=0 to 0.1s (burst) Producer sends 100 jobs : 3 workers pick up jobs 1-3 : Queue holds 97 (backpressure) section t=0.1s onward (drain) Workers pull from queue : Queue drains steadily : Bounded queue slows producer if full

Backpressure: If workers can't keep up, the queue fills. A bounded queue (maxsize=100) will make await queue.put() block, naturally slowing the producer. No data loss, no lost events.

The Code: Complete Example

Pattern: Async workers all calling await queue.get(). Only one gets each job. Use queue.join() to know when all jobs are done.

import asyncio

NUM_WORKERS = 3
SENTINEL = None  # signal to stop

async def process(job):
    """Simulate async work: DB query, API call, etc."""
    await asyncio.sleep(0.5)
    print(f"  ✓ job {job['id']}")

async def worker(name, queue):
    while True:
        job = await queue.get()  # ← blocks here if queue is empty
        try:
            if job is SENTINEL:
                return  # shutdown signal
            print(f"{name} picked up job {job['id']}")
            await process(job)
        finally:
            queue.task_done()  # notify queue that this job is done

async def producer(queue):
    for i in range(10):
        await queue.put({"id": i})
        await asyncio.sleep(0.1)

async def main():
    queue = asyncio.Queue(maxsize=5)  # bounded → backpressure

    async with asyncio.TaskGroup() as tg:
        # Start workers
        for n in range(NUM_WORKERS):
            tg.create_task(worker(f"worker-{n}", queue))

        # Produce jobs
        await producer(queue)

        # Wait for all jobs to finish
        await queue.join()

        # Tell workers to stop
        for _ in range(NUM_WORKERS):
            await queue.put(SENTINEL)

    print("done!")

asyncio.run(main())

Key Methods Explained

await queue.get()

Suspends the coroutine until an item is available. Safe for concurrent access.

await queue.put(item)

Adds an item. Blocks if queue is full (backpressure).

queue.task_done()

Decrements the internal counter. Call after finishing a job.

await queue.join()

Waits until all jobs are done (counter = 0).

Real-World: Webhook Processing

# Webhook handler (FastAPI)
@app.post("/webhook")
async def handle_webhook(event: dict):
    await webhook_queue.put(event)  # ← enqueue and return immediately
    return {"status": "queued"}

# Background worker
async def webhook_processor():
    while True:
        event = await webhook_queue.get()
        try:
            await process_stripe_webhook(event)  # network I/O
        finally:
            webhook_queue.task_done()

# Start worker on app startup
# Creates a bounded queue that slows webhook submissions if workers fall behind
webhook_queue = asyncio.Queue(maxsize=1000)

Key Takeaways

Decoupling

Producer doesn't wait for worker; it enqueues and continues. Workers pull from queue at their own pace.

Natural Backpressure

Bounded queue fills if workers slow down, which blocks producer. No data loss, handles bursts.

Scalability

Add more workers to handle faster processing. Queue automatically distributes jobs fairly.

Lifecycle

Use SENTINEL to shut down workers cleanly. Use join() to know when done.