asyncio.Queue.join зависает: где вызывать task_done

Работник взял элемент из asyncio.Queue и упал на неправильном задании. В очереди уже пусто, но await queue.join() не завершается. Пустота контейнера и окончание работы — разные состояния.

Понять счётчик незавершённых элементов

Каждый put увеличивает счётчик незавершённых заданий. get извлекает объект, но счётчик не уменьшает: работник ещё только начал обработку. task_done сообщает об окончании одной попытки обработки. join ждёт, когда для каждого добавленного элемента придёт такое сообщение.

Поэтому task_done удобно поставить в finally сразу после успешного get. Тогда ожидаемая ошибка обработки тоже завершает учёт элемента. Это не объявляет бизнес-операцию успешной: успех и отказ сохраняются отдельно. Если нужна повторная попытка, она должна явно добавить новое задание или изменить другой учётный механизм.

Не потерять ошибочное задание

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

Ожидаемые ValueError и TypeError обрабатываются внутри задания. Неожиданная ошибка программы не должна молча превращаться в обычный статус. В рабочем исполнителе владелец задач должен также замечать падение самого работника: один join не заменяет ожидание его Task.

import asyncio

async def main():
    queue = asyncio.Queue()
    results, rejected = [], []
    async def worker():
        while True:
            item = await queue.get()
            try:
                if item is None:
                    return
                try:
                    results.append(int(item) * 2)
                except (ValueError, TypeError):
                    rejected.append(item)
            finally:
                queue.task_done()
    task = asyncio.create_task(worker())
    for item in ["3", "bad", "4", None]:
        await queue.put(item)
    await asyncio.wait_for(queue.join(), timeout=2)
    await task
    assert results == [6, 8]
    assert rejected == ["bad"]
    assert queue.empty()
    print(results, rejected)

asyncio.run(main())

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

Проследите каждый путь после get: нормальная обработка, известная ошибка, сигнал остановки. Для каждого должен быть ровно один task_done. Лишний вызов приведёт к ValueError, пропущенный оставит join в ожидании. Не добавляйте task_done в нескольких ветках, если один finally может выразить ту же гарантию.

Отмена работника посреди задания тоже выполнит finally. В таком случае очередь посчитает попытку завершённой, но результат может отсутствовать. Если требуется восстановление после отмены, сохраняйте статус задания отдельно. Счётчик Queue живёт в памяти процесса и не является журналом доставки.

Начать с проверяемого результата

В публичном уроке pytest потренируйте обычный вход и отказ. В своей очереди проверьте их совместный запуск, как в примере. Откройте программу Python, чтобы продолжить с изоляцией состояния и регрессионными тестами.