asyncio в Python: корутины, задачи и цикл событий

FastAPI, aiohttp, асинхронные драйверы БД и боты построены на asyncio, поэтому вопросы о нём стали обязательными для бэкенда. Главное, что надо понимать: asyncio это кооперативная многозадачность в одном потоке.

Корутина, задача, цикл событий

import asyncio

async def fetch(n):
    await asyncio.sleep(1)    # здесь корутина уступает управление
    return n * 2

async def main():
    result = await fetch(1)   # последовательно, 1 секунда

asyncio.run(main())

• async def создаёт функцию-корутину. Её вызов fetch(1) ничего не выполняет, а возвращает объект корутины.
• await выполняет корутину и ждёт результат. Пока ожидаемая операция не готова, цикл событий выполняет другие задачи.
• Цикл событий (event loop) держит очередь готовых задач и следит за сокетами и таймерами.
• Задача (asyncio.Task) оборачивает корутину и планирует её выполнение конкурентно.

asyncio.run() создаёт цикл, выполняет главную корутину и закрывает цикл. Вызывается один раз на входе программы.

Конкурентность: gather и TaskGroup

await по очереди не даёт параллельности. Чтобы ждать одновременно, создают задачи:

async def main():
    results = await asyncio.gather(fetch(1), fetch(2), fetch(3))   # ~1 секунда

    async with asyncio.TaskGroup() as tg:                           # 3.11+
        t1 = tg.create_task(fetch(1))
        t2 = tg.create_task(fetch(2))
    print(t1.result(), t2.result())

• gather возвращает результаты в порядке аргументов. При ошибке в одной задаче по умолчанию бросает её, но остальные задачи не отменяются и продолжают работать. С return_exceptions=True вернёт исключения в списке.
• TaskGroup это структурированная конкурентность: если одна задача упала, остальные отменяются, а ошибки собираются в ExceptionGroup. Выход из блока гарантирует, что все задачи завершены.

asyncio.create_task(coro) запускает задачу «в фоне». Обязательно храните ссылку на неё: цикл держит только слабую ссылку, и задачу без ссылок может собрать сборщик мусора.

Таймауты и отмена

async with asyncio.timeout(2):       # 3.11+
    data = await fetch_slow()
# или
data = await asyncio.wait_for(fetch_slow(), timeout=2)

Отмена (task.cancel()) бросает CancelledError в точке await внутри задачи. Его можно поймать для очистки, но нужно пробросить дальше. CancelledError наследуется от BaseException, поэтому except Exception его не глотает.

Блокирующий код убивает цикл

Всё работает в одном потоке. Если корутина вызывает time.sleep(5), requests.get() или считает что-то тяжёлое, весь цикл стоит: ни одна другая задача не выполняется.

• Для ожидания: await asyncio.sleep().
• Для синхронных библиотек: await asyncio.to_thread(func, *args) (3.9+) или loop.run_in_executor.
• Для CPU-bound: пул процессов.
• Для HTTP и БД: асинхронные клиенты (httpx, aiohttp, asyncpg).

Ограничение параллелизма

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:
        return await client.get(url)

Без ограничения gather по 10 000 URL откроет 10 000 соединений разом.

Синхронизация

В asyncio свои примитивы: asyncio.Lock, Event, Queue, Semaphore. Они не потокобезопасны и работают только внутри одного цикла. Гонки возможны и здесь: между двумя await другая задача может изменить общие данные.

Частые ошибки

• Вызов корутины без await: «coroutine was never awaited».
• Блокирующий вызов внутри async def.
• create_task без сохранения ссылки и без обработки ошибок.
• Последовательные await там, где нужен gather.
• Глотание CancelledError.

Что спросят на собесе

• Чем asyncio отличается от потоков, где здесь параллельность.
• Что такое корутина, задача, цикл событий.
• Как выполнить запросы конкурентно, чем gather отличается от TaskGroup.
• Что будет, если в корутине вызвать time.sleep или requests.get, как это исправить.
• Как сделать таймаут и как работает отмена.