Producer-consumer and bounded queues

Decoupling threads with a queue that blocks when full or empty.

Low-level design conceptConcurrency in one process2 Nov 2026

01The intuition

The problem it solves

A photo app lets users upload pictures; each upload must be resized, which takes two seconds. The first version resizes inside the upload handler, so the user waits, and twenty uploads at once mean twenty resizes at once and a machine on its knees.

def handle_upload(image: bytes) -> None:
    thumb = resize(image)       # 2 seconds, on the request thread
    save(thumb)

def resize(image: bytes) -> bytes:
    return image

def save(thumb: bytes) -> None:
    pass

Two different speeds are coupled: uploads arrive in bursts, resizing happens at a steady pace. Whenever a fast producer meets a slow consumer, the fix has the same shape: put a queue between them, and make the queue refuse to grow without limit.

Build it one piece at a time

Step one: a queue and a worker thread. The handler appends the image and returns at once; one worker takes images off the queue and resizes them. The user waits milliseconds, and resizes happen one at a time.

import threading
from queue import Queue

jobs: Queue[bytes] = Queue()

def handle_upload(image: bytes) -> None:
    jobs.put(image)             # returns immediately

def worker() -> None:
    while True:
        image = jobs.get()      # blocks until an item exists
        save(resize(image))
        jobs.task_done()

threading.Thread(target=worker, daemon=True).start()

Step two: bound the queue. If uploads outrun resizing for an hour, an unbounded queue eats all memory. A bounded queue makes put block (or fail fast) when it is full, which pushes the pressure back to the producer; this is backpressure, and it is the point of the pattern, not a side effect.

from queue import Full

jobs = Queue(maxsize=100)

def handle_upload(image: bytes) -> bool:
    try:
        jobs.put(image, timeout=0.5)    # wait briefly, then refuse
        return True
    except Full:
        return False                    # tell the user to retry later

Step three: several consumers, and a clean shutdown. More workers consume faster on a multi-core machine; a sentinel value tells each worker to stop, and join waits until the queue is drained. Wrap all of it in a class so the thread count and the queue size are visible decisions.

class ResizePool:
    STOP = None

    def __init__(self, workers: int, capacity: int) -> None:
        self.jobs: Queue = Queue(maxsize=capacity)
        self.threads = [threading.Thread(target=self._run) for _ in range(workers)]
        for t in self.threads:
            t.start()

    def submit(self, image: bytes, timeout: float = 0.5) -> bool:
        try:
            self.jobs.put(image, timeout=timeout); return True
        except Full:
            return False

    def _run(self) -> None:
        while (image := self.jobs.get()) is not self.STOP:
            save(resize(image))

    def shutdown(self) -> None:
        for _ in self.threads:
            self.jobs.put(self.STOP)
        for t in self.threads:
            t.join()

Queue occupancy with capacity 3, one worker that takes 2 s per job, and five uploads at t = 0:

t (s)	Queue	Worker	Fifth upload
0	[img2, img3, img4]	img1	blocked on put, queue full
0.5	[img2, img3, img4]	img1	timeout, returns False
2	[img3, img4]	img2	would succeed now

Why it works

The queue's invariant is that every item put is taken exactly once, in order, and that the count never exceeds the capacity. The standard library's Queue holds a lock and two condition variables internally: consumers wait on not-empty, producers wait on not-full, and each put or get signals the other side. Because the handoff is the only shared state, the producer and consumer code need no locks of their own.

Where it breaks: a worker that raises dies silently and the pool shrinks; wrap the body in try-except and log. Work that must not be lost on a crash needs persistence, which an in-memory queue does not give. And a sentinel per worker is the only clean stop; killing threads is not available in Python and is unsafe elsewhere.

steady state: throughput = workers / service time. Queue full when arrival rate × service time > workers for longer than capacity allows

The trade-off you must say out loud

The user no longer sees the result immediately, so the API changes from "done" to "accepted"; that is a product decision, not just a code one. The capacity is a knob: too small and bursts are refused, too large and memory and latency grow. Say what happens when the queue is full (block, drop, or reject with a message) because that is the design. The follow-up that usually comes next is priorities, which turns the FIFO queue into a priority queue, and ordering, which a pool of workers no longer guarantees.

Where it shows up in the 100 Google problems

Blocking connection pool: acquirers are consumers of a bounded queue of connections; release is the producer side.
Concurrent priority task scheduler: the same pool with a priority queue and cancellation.
Logging framework: log calls enqueue, a single writer thread drains to sinks, so logging never blocks the caller.
Notification service: sends are queued per channel with retries handled by the consumer.
Coffee machine: orders queue in front of a few dispensers, the classic bounded buffer.