Queues

Contents

Queues#

For producer-consumer between threads, processes, or async tasks, use the right queue. Four flavours, picked by the concurrency model.

Use case

Module / type

Single-thread, double-ended

collections.deque

Thread-safe FIFO

queue.Queue

Thread-safe LIFO (stack)

queue.LifoQueue

Thread-safe priority

queue.PriorityQueue

Process-safe

multiprocessing.Queue

Async-coroutine-safe

asyncio.Queue

The standard producer-consumer structure.

import queue, threading

q = queue.Queue(maxsize=100)

def producer():
    for item in source:
        q.put(item)
    q.put(None)               # sentinel

def consumer():
    while (item := q.get()) is not None:
        process(item)
        q.task_done()

maxsize provides backpressure; the producer blocks on put when the queue is full.

For async, the structure is identical with await on get and put.

import asyncio
q: asyncio.Queue[Item] = asyncio.Queue(maxsize=100)

async def producer():
    async for item in stream:
        await q.put(item)
    await q.put(None)

async def consumer():
    while (item := await q.get()) is not None:
        await process(item)
        q.task_done()

References#

Contents