Если производитель создаёт задачи быстрее обработчика, задайте предел через asyncio.Queue(maxsize=2). При полном буфере await queue.put(item) ждёт свободного места. В примере Mentoring.digital одна корутина добавляет строки, другая читает их из очереди и завершает работу по специальному сообщению.
Почему пустая очередь и завершённые задачи различаются
После get() элемент уже убран из очереди, хотя обработка может ещё продолжаться. task_done() отмечает её завершение, а join() ждёт, пока отметка будет сделана для каждого добавленного элемента. Вызывайте task_done() ровно один раз после успешного получения.
Пример с одним обработчиком
Здесь None означает конец потока. Обработчик отмечает и это сообщение, иначе join() продолжит ждать. TaskGroup доступен начиная с Python 3.11.
import asyncio
async def consume(queue):
while True:
name = await queue.get()
try:
if name is None:
return
print(name.upper())
finally:
queue.task_done()
async def main():
queue = asyncio.Queue(maxsize=2)
async with asyncio.TaskGroup() as group:
group.create_task(consume(queue))
for name in ("alpha", "beta", "gamma"):
await queue.put(name)
await queue.put(None)
await queue.join()
asyncio.run(main())Что проверить при переносе в приложение
Запустите вариант с maxsize=1 и медленной обработкой. Следите за порядком событий, чтобы увидеть ожидание при заполнении. С несколькими обработчиками передайте по одному сообщению завершения каждому после остановки всех производителей.
Для задач с сетевыми вызовами заранее решите, где обрабатывается ошибка и нужна ли повторная попытка. Отметка в очереди завершает учёт элемента, поэтому успешный результат операции проверяйте отдельно. Сценарий с производителем и обработчиками подходит для практики асинхронного Python, а поведение при сбое можно обсудить с наставником по асинхронному коду на Python.
