asyncio.Queue en Python: Productor y Consumidor

asyncio.Queue conecta productores y consumidores dentro de un event loop. Con maxsize, un productor rápido espera cuando el backlog alcanza el límite, evitando crecimiento descontrolado de memoria.

import asyncio

async def consumidor(fila: asyncio.Queue[int | None]) -> None:
    while (item := await fila.get()) is not None:
        try:
            await asyncio.sleep(0.05)
            print(item)
        finally:
            fila.task_done()
    fila.task_done()

async def main() -> None:
    fila: asyncio.Queue[int | None] = asyncio.Queue(maxsize=10)
    worker = asyncio.create_task(consumidor(fila))
    for item in range(5):
        await fila.put(item)
    await fila.put(None)
    await fila.join()
    await worker

asyncio.run(main())

Cada elemento retirado con get() necesita task_done(). El centinela también cuenta y debe marcarse. join() no detiene consumidores; solo espera a que el trabajo pendiente llegue a cero.

Buenas prácticas

Usa await con put() y get() para respetar el backpressure. Los métodos no reciben timeout directamente, así que envuelve la operación con la herramienta adecuada de tu versión. Define políticas para fallos, cancelación y reintentos.

Usa asyncio.TaskGroup para supervisar consumidores. Si los productores están en threads, elige queue.Queue.

La documentación oficial de asyncio.Queue, consultada el 22 de julio de 2026, detalla la API y sus límites.