TopolearnQueues give you bounded concurrency and back-pressure for crawlers, pipelines and job runners.
asyncio.Queue is a FIFO for coroutines (not thread-safe; use queue.Queue across threads). await q.put(item) waits if the queue is full (when created with maxsize), and await q.get() waits until an item is available. That waiting gives back-pressure: a fast producer slows down to match the consumers.
The standard pattern is a fixed pool of worker tasks:
- start N workers that loop:
item = await q.get(), process,q.task_done(), - the producer puts items,
await q.join()waits until every put item has been marked done.
Then the workers must be stopped. Before 3.13 you cancelled them or sent sentinel values. Python 3.13 adds `q.shutdown()`: after it, put() raises QueueShutDown, and get() raises QueueShutDown once the queue is empty, so workers can exit their loop cleanly. shutdown(immediate=True) also discards pending items.
PriorityQueue and LifoQueue are variants. Use the _nowait methods (put_nowait, get_nowait) when you are not allowed to wait, and handle QueueFull / QueueEmpty.
The number of workers is your concurrency limit, which is often clearer than a Semaphore.
Example
import asyncio
async def worker(name, q, results):
while True:
try:
n = await q.get()
except asyncio.QueueShutDown: # 3.13+
return
await asyncio.sleep(0.01)
results.append(n * n)
q.task_done()
async def main():
q, results = asyncio.Queue(maxsize=3), []
async with asyncio.TaskGroup() as tg:
for i in range(2):
tg.create_task(worker(f"w{i}", q, results))
for n in range(6):
await q.put(n) # waits while the queue is full
await q.join() # all items processed
q.shutdown() # workers exit their loops
print(sorted(results)) # [0, 1, 4, 9, 16, 25]
asyncio.run(main())Remember
- put() waits when full, get() waits when empty: built-in back-pressure
- Pair every get() with task_done() so join() can finish
- 3.13's Queue.shutdown() stops workers without sentinels
Watch out for
- Forgetting task_done(), so await q.join() hangs forever.