Suppose you need to generate thumbnails for 50,000 uploaded images, calling an image service that can handle about ten requests at a time. asyncio.gather over 50,000 coroutines would create them all at once, open far too many connections and hold every result in memory. A plain for loop with await would take hours. What you want sits in between: a fixed number of workers pulling jobs from a queue, and a producer that slows down when the workers fall behind.
That’s a bounded worker queue, and asyncio.Queue gives you every piece of it. The pieces are simple, but the wiring has three places where a mistake makes the program hang or quietly lose work. This guide builds the pattern, runs it, and then breaks it on purpose so you can recognise the failures. Output is from Python 3.14; everything except the final shutdown section works on 3.8 and later.
The parts and what each one controls
There are four moving parts:
- The queue, created with
asyncio.Queue(maxsize=N). It holds jobs that are waiting for a worker. - The producer, which calls
await queue.put(job)for each job. - The workers, each a long-running task that loops forever:
await queue.get(), process the job, callqueue.task_done(). - The coordinator, which starts the workers, runs the producer, waits with
await queue.join()until every job is processed, and then stops the workers.
Two numbers shape the behaviour, and it’s worth being precise about which controls what. The number of workers is your concurrency limit: with ten workers, at most ten jobs are being processed at once, because each worker handles one job at a time. maxsize limits how many jobs can wait in the queue. It doesn’t change how fast work gets done; it limits how far the producer can run ahead and how much memory the waiting jobs use.
Why maxsize matters
With the default maxsize=0, the queue is unbounded and put never waits. A producer reading a 50,000-line CSV will load all 50,000 jobs into memory in a fraction of a second, long before the workers have finished the first batch:
q = asyncio.Queue()
for i in range(100_000):
await q.put(i)
print("unbounded queue size:", q.qsize()) # unbounded queue size: 100000That’s fine for integers, but if each job holds a downloaded file or a database row, you’ve just moved your entire input into RAM. With a maxsize, put suspends the producer while the queue is full and resumes it when a worker takes something off. That automatic slowing of the producer is called backpressure, and it’s the main reason to use a bounded queue at all.
If the producer must not wait (it’s handling incoming requests, for example), put_nowait raises asyncio.QueueFull instead of suspending, and you can reject or shed the job with a “try later” response.
A working pipeline, step by step
Here’s the thumbnail job scaled down: ten images, two workers, a queue that holds three, and a fake resize that takes 0.3 seconds and fails on image 7:
import asyncio, time
T0 = time.perf_counter()
def t():
return f"{time.perf_counter() - T0:4.1f}s"
async def resize(image_id):
await asyncio.sleep(0.3)
if image_id == 7:
raise ValueError(f"corrupt image {image_id}")
return f"thumb-{image_id}"
async def producer(queue, n):
for i in range(n):
await queue.put(i)
print(f"{t()} queued {i:2} (size {queue.qsize()})")
async def worker(name, queue, results, errors):
while True:
image_id = await queue.get()
try:
results.append(await resize(image_id))
except Exception as e:
errors.append((image_id, repr(e)))
finally:
queue.task_done()
async def main():
queue = asyncio.Queue(maxsize=3)
results, errors = [], []
workers = [asyncio.create_task(worker(f"w{i}", queue, results, errors))
for i in range(2)]
await producer(queue, 10)
print(f"{t()} producer finished, waiting for queue to drain")
await queue.join()
for w in workers:
w.cancel()
await asyncio.gather(*workers, return_exceptions=True)
print(f"{t()} done: {len(results)} ok, errors={errors}")
asyncio.run(main()) 0.0s queued 0 (size 1)
0.0s queued 1 (size 2)
0.0s queued 2 (size 3)
0.0s queued 3 (size 2)
0.0s queued 4 (size 3)
0.3s queued 5 (size 2)
0.3s queued 6 (size 3)
0.6s queued 7 (size 2)
0.6s queued 8 (size 3)
0.9s queued 9 (size 2)
0.9s producer finished, waiting for queue to drain
1.5s done: 9 ok, errors=[(7, "ValueError('corrupt image 7')")]Read the timestamps. The first three puts fill the queue instantly. At put 3 the size drops back to 2, because the workers have started and each took one job off. After job 4 the queue is full and both workers are busy, so the producer is suspended. It wakes at 0.3 seconds, when both workers finish and take the next two jobs. From then on the producer moves in steps of two jobs every 0.3 seconds, exactly matching the workers’ pace. That’s backpressure working.
The total is 1.5 seconds: ten jobs, two at a time, 0.3 seconds each. Doubling maxsize wouldn’t change that; adding workers would, up to whatever the image service can handle.
How task_done and join fit together
The queue keeps a counter of unfinished jobs. Every put increments it. Every task_done decrements it. join() waits until the counter reaches zero. So join means “every job that was put has been reported as finished”, which isn’t the same as “the queue is empty”. When the producer finishes at 0.9 seconds, the queue still holds jobs, and two more are being processed. join correctly waits for all of them.
get() doesn’t decrement anything. If you forget task_done, join waits forever.
Errors belong to the item, not the worker
The try/except/finally in the worker is the most important part of the code. Image 7 raised ValueError, the worker recorded it and moved on to the next job, and task_done ran in finally regardless. One bad input cost one result, not the pipeline.
Here’s what happens without that structure:
async def leaky_worker(q):
while True:
i = await q.get()
await resize(i) # raises for 7: the task dies here
q.task_done()Running the same ten jobs through two of these, with a 3-second timeout on join:
join() still waiting after 3s; unfinished tasks: 1 | dead workers: 1The worker that picked up job 7 crashed with the exception, so task_done never ran for that job. The other worker processed the rest, but the counter is stuck at one, and join never returns. It looks like a job stuck at 99% with no error in the logs, since the exception sits in a task nobody awaits. If enough jobs fail, every worker dies and the producer then blocks forever on a full queue, which is even harder to diagnose.
Catch Exception, not BaseException. CancelledError is a BaseException since Python 3.8, so catching Exception lets cancellation pass through and stop the worker during shutdown, which is what you want.
Shutting the workers down cleanly
Workers loop forever, so something has to stop them. Two approaches are common.
Cancel and await is what the main example uses. After join returns, every job is done and each worker is suspended in queue.get(). Cancelling them raises CancelledError at that await, and gather(*workers, return_exceptions=True) waits until each has actually finished, collecting the CancelledErrors rather than raising them. Skipping that final gather leaves the tasks half-cancelled when your function returns.
Sentinels were the usual alternative before 3.13: the producer puts one None per worker after the real jobs, and a worker returns when it receives None. It works, but you need exactly one sentinel per worker, and they must be put after all real jobs.
Queue.shutdown in Python 3.13+
Python 3.13 added Queue.shutdown(), which replaces sentinels. After shutdown(), new put calls raise asyncio.QueueShutDown; get continues returning the remaining items and raises QueueShutDown once the queue is empty. Combined with a TaskGroup, the whole lifecycle fits in one block:
async def worker():
while True:
try:
i = await q.get()
except asyncio.QueueShutDown:
return
try:
done.append(await resize(i))
except Exception:
pass # record the error in real code
finally:
q.task_done()
async with asyncio.TaskGroup() as tg:
for _ in range(2):
tg.create_task(worker())
for i in range(10):
await q.put(i)
q.shutdown()The TaskGroup block exits once both workers return, by which point all ten jobs have been handled (nine succeeded, matching the earlier run). Calling q.shutdown(immediate=True) discards the remaining items instead, which is useful for an emergency stop.
Sizing it in practice
Choose the worker count from the downstream limit, not from CPU cores. Asyncio workers mostly wait on I/O, so the right number is whatever the image service, database pool or API rate limit allows: if the service allows ten concurrent requests, use ten workers.
Set maxsize to a small multiple of the worker count, two to four times is typical. That’s enough buffered work that a worker never sits idle waiting for the producer, and small enough that memory stays flat. Work that’s CPU-heavy, like real image resizing with Pillow, shouldn’t run inside the coroutine at all, since it blocks the event loop; hand it to asyncio.to_thread or a process pool from within the worker.
Interview exercise: a crawler that never finishes
Question. A web crawler uses an asyncio.Queue(maxsize=100) and 20 workers. Each worker fetches a URL, parses the page, and calls await queue.put(link) for every new link it finds. The coordinator seeds the queue with one URL and awaits queue.join(). On a small site it works; on a large site it freezes with all workers idle and nothing in the logs. What’s going on?
Answer. The workers are also producers, and the queue is bounded. A worker that finds 150 links on a page tries to put them all, the queue fills at 100, and the worker blocks in put, before it reaches task_done for the page it’s processing. Once all 20 workers are blocked in put at the same time, nobody is left to call get, so the queue never drains. That’s a deadlock: every worker is waiting for space that only another worker could free. The silence in the logs is expected, because nothing failed. Fixes include making the frontier unbounded (with a separate limit on total pages crawled so memory stays bounded) or using put_nowait and keeping overflow links in a set or on disk. A strong answer also keeps a seen set so the same URL isn’t queued twice (which otherwise makes the crawl endless) and wraps each page in try/finally around task_done, so one failed fetch doesn’t leave join waiting.