Как устроен asyncio.Queue и как его использовать для паттерна producer-consumer в асинхронном коде?

Короткий ответ

asyncio.Queue — асинхронная FIFO-очередь с неблокирующим put/get: producer-корутины добавляют элементы через await queue.put(item), а consumer-корутины забирают их через await queue.get(), не блокируя event loop во время ожидания.

Как это работает подробнее

Методы put/get — корутины, которые корректно встраиваются в event loop: ожидание места в очереди или появления элемента не блокирует процесс, а отдаёт управление другим задачам. task_done() и join() позволяют дождаться, пока все поставленные элементы будут обработаны.

Пример кода

example-272.pyКопировать

import asyncio
async def producer(queue, n):
    for i in range(n):
        await queue.put(i)
    await queue.put(None)  # сигнал окончания для consumer
async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"обработан: {item}")
async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(producer(queue, 10), consumer(queue))
asyncio.run(main())
# Проверка: все 10 элементов будут выведены по порядку, затем корутины корректно завершатся

Что использовать на практике

• Практика. Паттерн применяют для ограничения конкурентности, буферизации между стадиями конвейера. Важно явно завершать consumer-ы (например, через сигнальное значение None или queue.join()), иначе они будут ждать вечно после того, как producer закончил работу.

Подводные камни

• Риски. Забывают про завершение consumer-корутин после обработки всех данных, из-за чего программа не завершается; используют обычный синхронный queue.Queue вместо asyncio.Queue в async-коде, что приводит к блокировке event loop.