Производитель быстрее потребителя: ограничиваем буфер asyncio.Queue

Приложение читает задания из потока и отправляет их медленному обработчику. Пока нагрузка мала, список ожидающих выглядит безобидно. Когда источник ускоряется, в памяти накапливаются сами задания и связанные с ними данные. Ограничение числа одновременно работающих обработчиков не ограничивает число уже созданных ожидающих задач.

Очередь с maxsize задаёт размер промежуточного буфера. Когда он заполнен, await queue.put ждёт, пока потребитель заберёт элемент. Ожидание замедляет производителя в той же цепочке исполнения. Так скорость чтения начинает зависеть от доступной ёмкости, а не от желания заранее поставить в очередь весь источник.

Очередь с ограниченной ёмкостью

Пример ниже делает границу видимой без измерения памяти. Очередь вмещает один элемент. Второй put запускается задачей и не завершается, пока первый элемент не извлечён. Event сообщает, что производитель уже дошёл до попытки записи. Это позволяет проверить ожидание без случайной большой задержки.

import asyncio

async def main():
    queue = asyncio.Queue(maxsize=1)
    started = asyncio.Event()
    await queue.put("first")

    async def produce():
        started.set()
        await queue.put("second")

    producer = asyncio.create_task(produce())
    await started.wait()
    assert not producer.done()
    first = await queue.get()
    await producer
    second = await queue.get()
    assert (first, second) == ("first", "second")
    print(first, second)

asyncio.run(main())

Где снова появляется накопление

Запуск Python 3.12 печатает first second. После started.wait задача производителя дошла до блокирующего put, но ещё не завершилась. get освобождает место, и второе значение попадает в очередь. Здесь проверяется именно вместимость буфера; учёт завершённой работы через task_done и join — другой контракт, его стоит проверять отдельно.

Чтобы ограничение работало в приложении, производитель должен действительно ожидать put. Если для каждого входного элемента создавать create_task(queue.put(item)), ожидающие coroutine сами станут неограниченным буфером. Значения уже прочитаны, задачи уже созданы, и maxsize регулирует только число элементов внутри Queue. Читайте источник последовательно либо используйте ограниченное число производителей.

Выбор maxsize зависит от размера элемента и допустимого времени ожидания. Сто тысяч маленьких ключей и сто тысяч загруженных документов требуют разной памяти. Если источник позволяет получать крупные страницы, ограничьте и размер страницы: очередь не контролирует буфер внутри клиента. В наблюдаемость полезно добавить время ожидания put, чтобы видеть отставание обработки.

При остановке заранее определите судьбу ожидающих производителей и оставшихся элементов. Нельзя просто забыть задачу, которая ждёт места: она останется живой. Управляющий код должен завершить производителей, дождаться нужной работы и остановить потребителей выбранным протоколом. Ограниченная очередь решает накопление между двумя этапами, когда оба этапа участвуют в этом протоколе.

Когда локальный пример работает, полезно разобрать соседние механизмы языка и их взаимодействие. В курсе «Python для профессионалов» есть практика по итераторам, генераторам, декораторам, asyncio, метапрограммированию и профилированию. Эти темы помогают понимать время жизни объектов, управление выполнением и причины поведения прикладной программы.

Посмотреть программу и практику курса.

Документация: официальный источник по теме.

Другие задачи по теме: Типы, тесты и поведение программы.