Основы async-кода в Python: как выбирать между create_task, gather и последовательным выполнением

Асинхронность в Python — не про «ускорить всё любой ценой», а про корректное управление ожиданиями (I/O) и временными зависимостями (когда результат нужен позже). Большинство проблем в реальных проектах возникает не из-за отсутствия async/await, а из-за неправильного выбора примитива: где-то задачи создаются «в никуда» и утекают, где-то вы теряете контроль конкурентности, а где-то исключения обрабатываются так, что вы или не узнаёте о сбоях, или узнаёте слишком поздно.

В этом материале разложим по полкам три наиболее частых сценария:

• последовательное выполнение await внутри цикла;
• конкурентное создание задач через asyncio.create_task;
• сбор результатов и управление исключениями через asyncio.gather.

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

Когда асинхронность реально нужна: модель ожиданий и зависимостей

Перед выбором create_task/gather полезно зафиксировать ментальную модель.

1. await не запускает параллельность сам по себе.
   Он просто «останавливает» текущую корутину и отдаёт управление event loop, пока awaited-объект (обычно — другая корутина/фьючер) завершится. Если вы последовательно делаете await в цикле, то операции будут идти по очереди (в рамках одной корутины).

2. Параллельность в asyncio — это конкуренция за event loop.
   То есть вы не получаете true CPU-параллельность. Асинхронность полезна при I/O: HTTP-запросы, чтение файлов, базы, ожидание таймеров, сокеты.

3. Создание задач (create_task) и сбор результатов (gather) — разные уровни управления.

   • create_task — вы явно говорите: «запусти это прямо сейчас как отдельную задачу».
   • gather — вы говорите: «собери результаты нескольких ожидаемых операций, и как-то обработай их завершение/ошибки».

В реальных ошибках почти всегда смешаны эти уровни: например, создаёте задачи, но не храните ссылки и не ждёте завершения; или используете gather, но ожидаете, что он остановит всё при первом исключении (а по умолчанию он может вести себя иначе, чем вы думаете).

Последовательное выполнение: когда это нормально и даже лучше

Самый простой паттерн:

results = []
for url in urls:
    data = await fetch(url)
    results.append(data)

Что хорошо

• Легко читать и отлаживать.
• Исключения всплывают естественно, останавливая текущую последовательность.
• Никаких утечек задач: вы ждёте каждую операцию до конца.

Что плохо

• Вы теряете конкурентность.
  Если fetch — сетевой I/O, то последовательная схема будет ждать завершения запроса по очереди.

Типичная ошибка: ожидать параллельности

Разработчики иногда пишут «асинхронный код» и считают, что он автоматически будет конкурентным. Нет: параллельность появляется только когда вы явно запускаете несколько корутин/задач и ждёте их совместно.

Когда последовательность — лучший выбор

• Есть строгая зависимость между шагами (например, нужно значение из первого запроса для формирования второго).
• Низкая нагрузка или требования по стабильности важнее скорости.
• Лимиты API/ресурсов: проще сделать по одному запросу, чем потом разбираться с rate limit и пиками нагрузки.

asyncio.create_task: управление жизненным циклом задач

create_task превращает корутину в объект Task, который планируется event loop независимо от текущей корутины.

task = asyncio.create_task(fetch(url))
result = await task

Ключевая идея

После create_task вы должны продумать:

• где хранить ссылку на задачу;
• кто и когда её await/отменит;
• как обработать исключения.

Если задача создаётся и ссылка теряется, а вы её не ждёте — исключения могут всплыть в логах как “Task exception was never retrieved”, а ресурсы могут остаться в неопределённом состоянии.

Утечки задач: самая частая проблема «as is, it works»

Утечка в asyncio часто выглядит так:

for url in urls:
    asyncio.create_task(fetch(url))  # создали, но не сохранили и не ждём
# функция завершается

Что может пойти не так

• Исключения не обрабатываются.
• Задачи продолжают работать после выхода из «родителя» (в зависимости от контекста event loop и того, как устроено приложение).
• Становится трудно объяснить поведение: результаты не собраны, порядок не гарантирован, диагностика усложняется.

Простой принцип

Если вы создаёте задачи — почти всегда нужно:

• хранить их в коллекции;
• await их завершение (или отменять при необходимости);
• ограничивать параллелизм (если задач может стать много).

Правильный шаблон конкурентного запуска: задачи + ожидание

Базовый безопасный вариант: создать список задач и затем дождаться их завершения.

tasks = [asyncio.create_task(fetch(url)) for url in urls]
results = [await task for task in tasks]  # порядок как у urls

Здесь есть нюанс: этот код ждёт задачи в порядке списка, но они уже запущены. То есть конкурентность есть, а порядок результатов сохраняется по исходному списку.

Но это не то же самое, что gather

При await task по очереди вы всё равно ждёте, но обработка исключений будет идти «как первый упавший await встретится в цикле». С точки зрения контроля исключений это может отличаться от gather.

Конкурентность и исключения: различия create_task/gather

Что делает gather

asyncio.gather(*aws) принимает ожидаемые объекты (обычно корутины) и возвращает единый awaitable.

• Он запускает все корутины (фактически — создаёт задачи внутри event loop).
• Возвращает список результатов в том же порядке, что и входные объекты.
• Имеет управляемое поведение по исключениям через параметр return_exceptions.

asyncio.gather: сбор результатов и управляемые сбои

Типичный вариант:

results = await asyncio.gather(*(fetch(url) for url in urls))

Поведение по умолчанию (return_exceptions=False)

Если одна из корутин выбрасывает исключение:

• gather завершится с исключением;
• остальные корутины могут быть отменены не всегда одинаково предсказуемо в зависимости от версии/контекста и того, как исключение прокидывается;
• вы можете получить каскад отмен и «не те» первичные причины, если не продумали обработку.

Важно: не стоит рассчитывать, что gather всегда ведёт себя «как хочется» без понимания параметров и контекста. На практике лучше явно выбрать стратегию: “падать на первом исключении” или “собирать всё и потом разбирать”.

gather(..., return_exceptions=True): режим “собрать всё, где возможно”

Часто вы хотите: запросов много, один упал — остальные пусть продолжат, а потом вы решите, что делать с ошибками.

results = await asyncio.gather(
    *(fetch(url) for url in urls),
    return_exceptions=True
)

ok = []
errors = []
for url, item in zip(urls, results):
    if isinstance(item, Exception):
        errors.append((url, item))
    else:
        ok.append(item)

Почему это полезно

• У вас появляется данные + ошибки в одном месте.
• Легче писать предсказуемую бизнес-логику (например, показать частичный результат пользователю или повторить только упавшие).

Подводный камень

• Исключения не «срывают» выполнение, поэтому возможна ситуация, когда вы забыли проверить типы результатов и дальше обработали исключение как будто это валидные данные.

Частая ошибка №1: ожидать, что gather «ограничит» конкурентность

gather не ограничивает количество одновременно выполняющихся задач. Если urls — это 100000 элементов, вы в один момент создадите огромный набор задач и упадёте по:

• памяти,
• дескрипторам сокетов,
• ограничениям удалённого сервиса,
• rate limits.

Правильный подход: лимит конкурентности через semaphore

sem = asyncio.Semaphore(20)

async def fetch_limited(url):
    async with sem:
        return await fetch(url)

results = await asyncio.gather(*(fetch_limited(url) for url in urls))

Теперь одновременно выполняется не более 20 fetch.

Ошибка №2: смешивание create_task и gather без ясной стратегии

Иногда встречается конструкция:

tasks = [asyncio.create_task(fetch(url)) for url in urls]
results = await asyncio.gather(*tasks)  # tasks уже задачи

Это не всегда «плохо», но важно понимать: gather принимает awaitables и работает и с задачами, и с корутинами. Однако вы теряете ясность: вы могли бы не создавать create_task вовсе и передать корутины напрямую — gather всё равно запустит их.

С точки зрения кода, предпочтительнее выбирать одну модель:

• или вы запускаете задачи вручную (create_task) и затем управляете жизненным циклом,
• или отдаёте управление gather (и тогда обычно не нужно create_task отдельно).

Ошибка №3: забыли обработать отмену (cancel) и оставили функции “висящими”

В асинхронном приложении отмена (task.cancel()) — нормальная часть контроля. Например, если пользователь закрыл страницу или истёк таймаут.

Если вы пишете корутины, которые делают I/O, и внутри нет корректной реакции на отмену, то могут быть “полу-сделанные” операции, блокировки и странные логи.

Минимальный паттерн для отмены — не заглатывать CancelledError случайно:

async def fetch(url):
    try:
        return await http_get(url)
    except asyncio.CancelledError:
        # отмена — это сигнал, его нужно пробросить
        raise

А вот “глотать” отмену как обычную ошибку — плохая идея: отмена теряет смысл, а отменяющий код считает, что всё остановилось.

Обработка исключений: три стратегии, которые стоит выбрать осознанно

В async-коде почти всегда возникает вопрос: что делать при ошибке в одной из операций?

Стратегия A: остановиться на первой ошибке

Подходит, когда ошибки критичны и частичный результат бессмысленен.

Пример: последовательный код естественно останавливается.

Для конкурентной версии можно использовать gather без return_exceptions=True:

try:
    results = await asyncio.gather(*(fetch(url) for url in urls))
except Exception as e:
    # первые существенные детали ошибки
    ...

Важно при этом всё равно понимать, что остальные операции могут быть в процессе; если вам нужно гарантированное завершение/отмена — делайте это явно (см. ниже).

Стратегия B: собрать все результаты и ошибки

Для аналитики, батч-обработки, UI “показать что смогли”.

Используйте:

results = await asyncio.gather(*(fetch(url) for url in urls), return_exceptions=True)

Стратегия C: явный контроль жизненного цикла задач

Когда нужно:

• отменять остальные при первом фатальном исключении,
• логировать причину и дождаться отмен,
• обеспечить детерминированность поведения.

Пример (более низкоуровневый, но предсказуемый подход):

tasks = [asyncio.create_task(fetch(url)) for url in urls]
try:
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
    for t in done:
        exc = t.exception()
        if exc is not None:
            # при первой ошибке отменяем остальные
            for p in pending:
                p.cancel()
            # ждём, чтобы отмена корректно завершилась
            await asyncio.gather(*pending, return_exceptions=True)
            raise exc

    # если ошибок не было, ждём оставшиеся
    results = [t.result() for t in done]
    if pending:
        results += await asyncio.gather(*pending)
    return results
except:
    # на всякий случай: гарантируем сбор исключений/отмен
    for t in tasks:
        t.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    raise

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

Порядок результатов: важная деталь для UX и бизнес-логики

• Последовательное выполнение возвращает результат в порядке обработки.
• gather возвращает результаты в порядке входных объектов, независимо от фактического времени завершения.
• При использовании create_task и последующего await task в цикле порядок будет таким же, как у списка задач, но если вы ждёте иначе (например, через asyncio.as_completed), порядок будет временем завершения.

Если порядок критичен — используйте gather либо сохраняйте соответствие с ключами.

Ограничение конкурентности: semaphore vs “свой пул задач”

Semaphore — самый простой и надёжный инструмент ограничения параллельности. Он работает и с gather, и с собственными задачами.

Для более сложных кейсов (например, при пакетной обработке или нестандартной отмене) можно строить пул и очередь, но начиная с Semaphore вы избежите множества проблем.

Типичная ошибка: создать N задач под urls и потом надеяться, что где-то “внутри” будет ограничение. Если ограничение не сделано явно — его нет.

Таймауты: сочетайте конкурентность и контроль времени

Асинхронные операции почти всегда должны иметь таймаут (либо на уровне HTTP-клиента, либо поверх корутины).

Правильная схема часто выглядит так:

async def fetch_with_timeout(url, timeout=5):
    return await asyncio.wait_for(fetch(url), timeout=timeout)

Дальше:

results = await asyncio.gather(
    *(fetch_with_timeout(url) for url in urls),
    return_exceptions=True
)

Если таймаут пробрасывается как исключение — в режиме return_exceptions=True он будет собран вместе с остальными ошибками. В режиме “падать на первом исключении” таймаут остановит gather.

Практическая шпаргалка: как выбирать между тремя подходами

Ниже — краткая, но рабочая логика выбора.

1) Нужно по шагам, результат каждого шага влияет на следующий

Выбирайте последовательное await.

2) Нужна конкурентность, но достаточно собрать результаты

Используйте asyncio.gather:

• если хотите падать на первой ошибке — по умолчанию;
• если хотите собрать всё — return_exceptions=True.

3) Нужен тонкий контроль отмены/ошибок/жизненного цикла

Используйте asyncio.create_task + asyncio.wait или собственные правила:

• отмена pending при первой фатальной ошибке;
• гарантированная корректная отмена и сбор исключений.

4) Есть риск перегрузить систему (1000+ операций)

Добавляйте лимит конкурентности через Semaphore независимо от того, gather или create_task.

Частые ошибки в реальных кодовых базах (и как их диагностировать)

“Task exception was never retrieved”

Причина: создана задача, но исключение не было получено через await/result().

Лечение:

• хранить задачи в списке и ждать;
• или использовать gather/wait так, чтобы все задачи были “обслужены”.

“Почему у нас растёт память?”

Причина: массовое создание задач без лимитов, особенно если операции зависают дольше ожидаемого или вы делаете батчи без контроля.

Лечение:

• Semaphore;
• таймауты;
• (по необходимости) backpressure (очереди/потоки событий).

“Почему порядок неправильный?”

Причина: вы использовали as_completed (или схожие механики) и обрабатывали результаты “по мере готовности”, но ожидали порядок входа.

Лечение:

• сохранять соответствие ключ→результат;
• или перейти на gather, где порядок входа сохраняется.

“Почему ошибка не та, или мы теряем первопричину?”

Причина: неправильная стратегия исключений: вы отменяете задачи, а дальше исключения затёрты или перехвачены повторно.

Лечение:

• определиться со стратегией: падать/собирать/отменять;
• в сложных случаях делать явный контроль задач (вариант с wait(FIRST_EXCEPTION)).

Заключение: дисциплина в асинхронности важнее «магии»

В Python async-код — это не