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

Используйте asyncio.as_completed. Он позволяет получать результаты задач в порядке их фактического завершения, поэтому быстрые результаты можно обработать или отправить клиенту сразу, не ожидая остальные.

В отличие от него, asyncio.gather возвращает результаты только после завершения всех переданных операций и сохраняет порядок исходного списка.

Исторический контекст

Асинхронное программирование появилось как способ эффективно обслуживать множество операций ввода-вывода без выделения отдельного потока на каждую операцию. Для этого нужны не только средства запуска задач, но и разные стратегии получения их результатов.

Ожидание всех результатов удобно для пакетной обработки. Однако в сетевых сервисах часто важнее реагировать на каждое завершение как можно раньше — именно такую модель поддерживает as_completed.

Постановка проблемы

Предположим, несколько корутин обращаются к независимым внешним сервисам. Если одна операция завершится за 50 миллисекунд, а другая — за 5 секунд, ожидание через gather задержит обработку быстрого результата до завершения самой медленной операции.

Это может увеличить задержку ответа, ухудшить потоковую выдачу данных и привести к бесполезному удержанию уже готовых результатов. Нельзя просто заменить gather на последовательный await: тогда операции начнут ожидаться по очереди, а не конкурентно.

Подробное решение

asyncio.as_completed принимает набор awaitable-объектов и возвращает итератор. Каждая следующая итерация предоставляет awaitable, соответствующий очередной завершившейся операции; после его ожидания можно получить результат или обработать исключение именно этой операции.

import asyncio

async def work(delay, value):
    await asyncio.sleep(delay)
    return value

async def main():
    tasks = [
        asyncio.create_task(work(0.3, "медленный")),
        asyncio.create_task(work(0.1, "быстрый")),
    ]
    for completed in asyncio.as_completed(tasks):
        try:
            print(await completed)
        except Exception as error:
            print(f"ошибка: {error}")

asyncio.run(main())

В примере сначала будет обработан результат операции с задержкой 0,1 секунды, хотя она находится второй в исходном списке. Само создание задач происходит до начала итерации, поэтому операции выполняются конкурентно в рамках одного event loop.

Исключения нужно обрабатывать вокруг await completed. Тайм-аут, переданный в as_completed, относится ко всему процессу ожидания: если все операции не успели завершиться за указанное время, ожидание завершается ошибкой тайм-аута. Уже выполняющиеся задачи при этом не обязательно автоматически прекращаются — их жизненным циклом нужно управлять отдельно.

У gather другая семантика: он агрегирует результаты в порядке исходных аргументов, а не в порядке завершения. Поэтому as_completed подходит для потоковой обработки, раннего отображения результатов и независимых операций, а gather — когда нужен единый итоговый набор результатов.

Ситуация из практики

Сервис параллельно запрашивает цены у нескольких поставщиков и хочет показывать пользователю предложения по мере их поступления. Вариант с последовательным await прост, но полностью теряет конкурентность. Вариант с gather сохраняет конкурентность, однако первый результат нельзя обработать до ответа самого медленного поставщика.

Выбран as_completed: каждое предложение обрабатывается сразу после завершения соответствующего запроса. Для каждого результата добавляется отдельная обработка исключений, а внешний тайм-аут ограничивает общее время ожидания. Если бизнес-логика допускает завершение после получения, например, первых трёх ответов, оставшиеся задачи дополнительно отменяются и корректно дожидаются завершения отмены.

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

Что кандидаты часто упускают

Сохраняет ли as_completed порядок исходного списка?

Нет. Он выдаёт операции по мере завершения. Если порядок важен, нужно либо использовать gather, либо самостоятельно сопоставлять результаты с исходными идентификаторами и сортировать их после обработки.

Отменяет ли as_completed остальные задачи после первой ошибки?

Сам по себе — нет. Ошибка конкретной операции становится видна при ожидании соответствующего элемента, но другие задачи могут продолжить выполнение. Если после ошибки требуется остановить остальные операции, их нужно явно отменить и затем дождаться их завершения, обычно через дополнительную обработку отмены.

Можно ли с помощью as_completed получить результат, не создавая Task заранее?

Да, в него можно передать корутины и другие awaitable-объекты; asyncio организует их выполнение. Однако явное создание Task часто удобнее, когда нужно позднее отменить операции, проверить их состояние или сопоставить каждую задачу с контекстом запроса.

Следует также учитывать версию Python: в новых версиях API поддерживает расширенный вариант асинхронной итерации, но обычный шаблон с for и последующим await остаётся распространённым способом использования.