asyncio в Python: корутины, async/await и запуск задач

asyncio выполняет много операций ожидания в одном потоке. Пока одна корутина ждёт ответа сети, цикл событий переключается на другую, поэтому пять запросов занимают примерно столько же времени, сколько один. Пишешь через async def и await, запускаешь через asyncio.run(). Выигрыш появляется только там, где код ждёт ввода-вывода: счёт на процессоре asyncio не ускоряет вообще.

Весь разбор идёт на одной задаче: сходить в пять эндпоинтов внутреннего API. Настоящей сети нет, задержку изображает asyncio.sleep — 200 миллисекунд на запрос. Всё считано на CPython 3.14.5 под Linux.

Что такое корутина и почему её вызов ничего не запускает

async def объявляет корутинную функцию. Её вызов возвращает объект корутины и на этом останавливается: тело не выполняется, asyncio.sleep не начинается, до return дело не доходит.

Ровно так же ведёт себя функция с yield, и это не совпадение: до появления async/await корутины писали генераторами, а приостановка на await — тот же замороженный кадр, что и на yield. Механика разобрана в статье про генераторы.

import asyncio
import time

ENDPOINTS = ["/users", "/orders", "/payments", "/profile", "/settings"]


async def fetch(url, delay=0.2):
    await asyncio.sleep(delay)
    return f"{url} 200"


fetch("/users")
print("конец")

main.py:12: RuntimeWarning: coroutine 'fetch' was never awaited
  fetch("/users")
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
конец

Обычная функция от вызова работает. Корутина только создаёт объект, который умеет работать. Здесь объект никто не забрал, сборщик мусора удалил его и напечатал предупреждение. Увидел coroutine ... was never awaited в логах, значит где-то потерян await.

Забрать корутину можно двумя способами: await ждёт результат прямо здесь, asyncio.create_task отдаёт её циклу и идёт дальше. Всё остальное сводится к этим двум. asyncio.gather, TaskGroup.create_task и asyncio.ensure_future сами заворачивают переданную корутину в задачу, поэтому им можно скармливать голые корутины. Все фрагменты ниже опираются на эти ENDPOINTS и fetch.

Как запустить: asyncio.run и цикл событий

asyncio.run() открывает всю конструкцию. Он создаёт цикл событий, выполняет переданную корутину до конца, закрывает цикл и возвращает результат. Функция появилась в Python 3.7.

async def main():
    start = time.perf_counter()
    for url in ENDPOINTS:
        print(await fetch(url))
    print(f"итого {time.perf_counter() - start:.2f} c")


asyncio.run(main())

/users 200
/orders 200
/payments 200
/profile 200
/settings 200
итого 1.00 c

Цикл событий устроен просто: очередь готовых продолжить задач и бесконечный while, который двигает их по очереди. Задача выполняется, пока не упрётся в await над чем-то незавершённым, и тогда возвращает управление циклу. Цикл берёт следующую готовую. Параллельности внутри одного цикла нет, в каждый момент работает ровно одна корутина, а выигрыш берётся из того, что ожидание ответа сервера процессор не занимает.

Отсюда ограничение: из уже работающего цикла asyncio.run() вызывать нельзя, получишь RuntimeError: asyncio.run() cannot be called from a running event loop. Внутри асинхронного кода используй await, а asyncio.run() оставь одному вызову на входе в программу.

Почему пять запросов идут секунду вместо 0.2

Потому что await означает «остановись здесь и дождись результата». Пять await подряд дают пять ожиданий подряд: 5 × 0.2 = 1 секунда, что замер выше и показал. Цикл событий на каждом await свободен, но занять его нечем, других задач в очереди нет. Асинхронный код сам по себе ничего не ускоряет, ускоряет одновременный запуск.

Отдай циклу все пять корутин сразу:

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(url) for url in ENDPOINTS))
    for line in results:
        print(line)
    print(f"итого {time.perf_counter() - start:.2f} c")

Те же пять строк результата, другое время:

итого 0.20 c

Все пять корутин упёрлись в свой asyncio.sleep почти одновременно, ждали тоже одновременно, общее время равно самому долгому запросу. Порядок результатов при этом соответствует порядку аргументов, а не тому, кто закончил первым.

create_task, gather и TaskGroup: что выбрать

asyncio.create_task(coro) отдаёт корутину циклу и сразу возвращает объект Task. Работать задача начнёт на ближайшем await в вызывающем коде, а не в момент создания.

async def main():
    start = time.perf_counter()
    task = asyncio.create_task(fetch("/orders", 0.3))
    print(f"{time.perf_counter() - start:.2f} c: задача создана, идём дальше")
    users = await fetch("/users", 0.1)
    print(f"{time.perf_counter() - start:.2f} c: {users}")
    orders = await task
    print(f"{time.perf_counter() - start:.2f} c: {orders}")

0.00 c: задача создана, идём дальше
0.10 c: /users 200
0.30 c: /orders 200

Задача /orders крутилась в фоне, пока main ждал /users. Финальный await task её не перезапускает, он забирает готовый результат.

asyncio.TaskGroup появился в Python 3.11. Это менеджер контекста, который на выходе из блока дожидается всех созданных внутри задач. На пяти запросах он даёт те же 0.20 c, что и gather. Настоящая разница между ними видна на ошибке: пусть /orders падает через 0.1 с, пока /users честно работает свои 0.2 с.

async def broken(url):
    await asyncio.sleep(0.1)
    raise RuntimeError(f"{url} 503")


async def with_gather():
    task = asyncio.create_task(fetch("/users"))
    try:
        await asyncio.gather(task, broken("/orders"))
    except RuntimeError as e:
        print("gather поймал:", e)
    await asyncio.sleep(0.3)
    print("  /users отменён:", task.cancelled())


async def with_taskgroup():
    try:
        async with asyncio.TaskGroup() as tg:
            task = tg.create_task(fetch("/users"))
            tg.create_task(broken("/orders"))
    except* RuntimeError as eg:
        print("TaskGroup поймал:", eg.exceptions[0])
    print("  /users отменён:", task.cancelled())

gather поймал: /orders 503
  /users отменён: False
TaskGroup поймал: /orders 503
  /users отменён: True

gather пробросил первую ошибку наверх, а /users остался работать в фоне и спокойно дошёл до конца. TaskGroup при падении любой задачи отменяет остальные и отдаёт ошибки группой, которую ловят через except*.

Выбор сводится к одному вопросу: что делать, когда одна задача упала. Если остальные результаты всё равно нужны, бери gather(..., return_exceptions=True), тогда исключения приедут в списке результатов вместо того, чтобы бросаться. Если без упавшей задачи вся пачка бессмысленна, это TaskGroup. А create_task остаётся для одиночной фоновой работы рядом с основной.

Блокирующий вызов останавливает весь цикл, а не одну задачу

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

Отчёт считается обычным time.sleep(0.5), два запроса ждут по 0.2 с через asyncio.sleep:

START = time.perf_counter()


def log(msg):
    print(f"{time.perf_counter() - START:.2f} c: {msg}")


async def timed(url):
    await fetch(url)
    log(f"{url} готов")


async def build_report():
    time.sleep(0.5)
    log("/report готов")


async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(build_report())
        tg.create_task(timed("/users"))
        tg.create_task(timed("/orders"))

0.50 c: /report готов
0.70 c: /users готов
0.70 c: /orders готов

Ожидаемые 0.50 превратились в 0.70. Запросы /users и /orders вообще не начинались, пока блокирующий вызов не отпустил поток: сначала 0.5 на блокировку, потом 0.2 на оба запроса. Одна тормозящая функция замедлила задачи, которые её даже не вызывают.

Лечится выносом синхронного кода в отдельный поток через asyncio.to_thread, доступный с Python 3.9:

def build_report_blocking():
    time.sleep(0.5)


async def build_report():
    await asyncio.to_thread(build_report_blocking)
    log("/report готов")

0.20 c: /users готов
0.20 c: /orders готов
0.50 c: /report готов

Запросы отработали за свои 0.2 с, общее время равно самой долгой операции. Список типовых блокировщиков короткий: time.sleep, requests и любой синхронный HTTP-клиент, драйвер БД без поддержки asyncio, чтение большого файла обычным open. Все они ждут снаружи интерпретатора и на время ожидания отпускают GIL, поэтому лечатся одинаково: to_thread или замена на асинхронный аналог.

Счёт на процессоре to_thread не лечит. Разбор 26 МБ JSON выполняет C-парсер, который GIL не отпускает: поток стартует, а цикл всё равно стоит. Замер на той же машине: to_thread(time.sleep, 1.0) держит задержку цикла на 0 мс, а to_thread(json.loads, raw) при разборе в 0.16 c морозит цикл на 156 мс, то есть почти на весь разбор. Такое выносят в процесс:

def parse_and_count(raw):
    return len(json.loads(raw))


async def build_report(pool, raw):
    loop = asyncio.get_running_loop()
    count = await loop.run_in_executor(pool, parse_and_count, raw)
    log(f"/report готов, записей {count}")

Разбор занял 0.21 c вместо 0.16 c, зато задержка цикла упала до 12 мс. Пул создают один раз на старте и передают внутрь: ProcessPoolExecutor() прямо в корутине сам поднимает процессы и снова тормозит цикл. Свёртку делает рабочий процесс, потому что аргумент и результат едут через pickle: вернуть число дешевле, чем список на 300 тысяч словарей.

Правило короткое: to_thread для ожидания, процессы для счёта.

Отмена: CancelledError нельзя молча глотать

task.cancel() не убивает задачу снаружи. Он помечает её на отмену и отменяет то, чего она сейчас ждёт, а CancelledError прилетит в корутину, когда управление вернётся циклу. Не в момент вызова. Поэтому task.cancel() вернёт True, а task.cancelled() следующей же строкой отдаст False: задача про отмену ещё не знает. Дальше всё решает сама корутина. С версии 3.8 этот класс наследуется от BaseException, поэтому issubclass(asyncio.CancelledError, Exception) возвращает False и привычный except Exception отмену не поймает.

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

async def polite(url):
    try:
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print(f"{url}: закрываю соединение")
        raise
    return f"{url} 200"


async def greedy(url):
    try:
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print(f"{url}: проглотил отмену")
    return f"{url} 200"

Каждую запускаем задачей, через 0.1 с зовём task.cancel(), затем печатаем итог await task и значение task.cancelled():

/users: закрываю соединение
  задача отменена
  task.cancelled() = True
/orders: проглотил отмену
  результат: /orders 200
  task.cancelled() = False

polite освободил ресурс и пробросил отмену: task.cancelled() вернул True, вызвавший код получил то, что просил. greedy перехватил CancelledError и продолжил работу, будто ничего не случилось. Задача вернула результат вместо отмены. Внутри TaskGroup или asyncio.timeout такая корутина откладывает выход из блока: группа её уже отменила, а она всё ещё работает.

Надёжнее вообще не ловить отмену, а положить уборку в finally. Тогда соединение закроется на любом исходе: при нормальном возврате, при ошибке и при отмене.

Когда asyncio не нужен

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

Счёт на процессоре. Четыре одинаковых расчёта sum(i * i for i in range(3_000_000)), сначала подряд, потом через gather:

последовательно: 0.39 c
через gather:    0.39 c

Абсолютные числа зависят от машины, важно их равенство. Цикл один, поток один, а корутина без await внутри держит управление до самого return. Для расчётов нужны процессы: ProcessPoolExecutor или multiprocessing. Почему потоки тут тоже не спасают, разобрано в статье про GIL.

Смешанный стек. asyncio требует асинхронных библиотек сверху донизу, и один синхронный драйвер базы в середине пути сводит выигрыш к нулю.

На чём ловят на собеседовании

Ссылку на фоновую задачу нужно хранить. Цикл событий держит на задачи только слабые ссылки, поэтому asyncio.create_task(...) без присваивания результата разрешает сборщику мусора удалить задачу в любой момент, даже недоделанную. Документация советует складывать такие задачи в множество и снимать оттуда по add_done_callback.

Ошибка в задаче, которую никто не ждал, теряется. Упало исключение, а await или .result() к задаче никто не применил, и программа завершится с кодом возврата 0. В поток ошибок улетит сообщение с трассировкой:

Task exception was never retrieved
future: <Task finished name='Task-2' coro=<broken() done, defined at main.py:4> exception=RuntimeError('/orders 503')>
Traceback (most recent call last):
  File "main.py", line 6, in broken
    raise RuntimeError(f"{url} 503")
RuntimeError: /orders 503

Место падения по логам найдётся, а вот мониторинг по коду возврата такое не заметит.

Корутину нельзя ждать дважды. Второй await над тем же объектом корутины даст RuntimeError: cannot reuse already awaited coroutine, объект одноразовый. Если результат нужен в нескольких местах, оберни корутину в задачу через create_task: у Task результат забирается сколько угодно раз.

Частые вопросы

Чем корутина отличается от потока

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

Как ограничить число одновременных запросов

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

async def limited(url, sem):
    async with sem:
        return await fetch(url)


async def main():
    sem = asyncio.Semaphore(2)
    await asyncio.gather(*(limited(url, sem) for url in ENDPOINTS))

Пять запросов по два за раз дают три волны и 0.60 c вместо 0.20 c. Потолок нужен почти всегда: без него тысяча URL превратится в тысячу одновременных соединений, и упрёшься либо в лимит файловых дескрипторов, либо в rate limit чужого API.

Почему requests не работает с asyncio

requests синхронный, он блокирует поток на всё время запроса, а вместе с потоком и весь цикл событий. Формально код выполнится, ускорения не будет. Варианта два: взять асинхронный клиент (aiohttp, httpx в async-режиме) либо оставить requests и звать его через asyncio.to_thread.

С какой версии Python это всё работает

Код из статьи требует 3.11 из-за TaskGroup. Всё остальное работает начиная с 3.9.

Что учить дальше

Соседний вопрос на любом собеседовании: чем asyncio отличается от потоков и процессов и что брать под конкретную задачу. Сравнение всех трёх подходов лежит в отдельном разборе.

Разобраться в отмене и в поведении групп задач проще руками. В паке «конкурентность» пути «Python: продвинутая практика» собраны ровно эти сюжеты:

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