TopolearnQueues replace most hand-written locking and give natural back-pressure between pipeline stages.
queue.Queue is a thread-safe FIFO: all locking is inside. It's the standard way for threads to communicate.
q.put(item)adds; withmaxsizeset, it blocks while the queue is full, which slows producers to the consumers' pace (back-pressure).q.get()blocks until an item is available.timeout=raisesqueue.Empty/queue.Full.- After processing each item, call
q.task_done();q.join()waits until every item put has been marked done.
Stopping consumers is the tricky part. Classic approaches put one sentinel (such as None) per worker. Python 3.13 adds `q.shutdown()`: further put() raises queue.ShutDown, and get() raises ShutDown once the queue is drained, so workers exit cleanly. shutdown(immediate=True) also discards remaining items.
Variants: LifoQueue (stack), PriorityQueue (smallest item first; use (priority, counter, item) tuples), and SimpleQueue (unbounded, no task_done, reentrant-safe).
These queues are for threads. Between processes use multiprocessing.Queue; between asyncio tasks use asyncio.Queue.
Example
import queue, threading
q: queue.Queue[int] = queue.Queue(maxsize=5)
results, lock = [], threading.Lock()
def worker():
while True:
try:
n = q.get()
except queue.ShutDown: # 3.13+: queue shut down and empty
return
with lock:
results.append(n * n)
q.task_done()
workers = [threading.Thread(target=worker) for _ in range(3)]
for w in workers: w.start()
for n in range(10):
q.put(n) # blocks while 5 items are waiting
q.join() # all items processed
q.shutdown() # wake workers so they exit
for w in workers: w.join()
print(sorted(results)) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]Remember
- queue.Queue is thread-safe; put/get block with optional timeouts
- task_done() + join() track completion
- Queue.shutdown() (3.13) stops consumers without sentinels
Watch out for
- Forgetting task_done(), so q.join() never returns.