asyncio: event loop, async/await, корутины

asyncio это библиотека стандартной библиотеки для асинхронного программирования. Она позволяет писать конкурентный код в одном потоке - сотни тысяч одновременных IO-операций без потоков. Это основа современных Python-веб-фреймворков (FastAPI, aiohttp) и любого backend, делающего много IO-запросов. В этом уроке - базовые понятия.

Зачем нужен asyncio

Представь сервер, который делает запросы к 100 микросервисам для обработки одного запроса пользователя. Синхронно:

def get_user_data(user_id):
    profile = fetch("/profile/" + user_id)       # 50ms
    posts = fetch("/posts?user=" + user_id)      # 50ms
    friends = fetch("/friends/" + user_id)       # 50ms
    return {"profile": profile, "posts": posts, "friends": friends}

# Время: 150ms (50 * 3)

С asyncio - параллельно:

async def get_user_data(user_id):
    profile, posts, friends = await asyncio.gather(
        fetch("/profile/" + user_id),
        fetch("/posts?user=" + user_id),
        fetch("/friends/" + user_id),
    )
    return {"profile": profile, "posts": posts, "friends": friends}

# Время: ~50ms (max из трёх) - параллельно

Запросы выполняются конкурентно. Программа ждёт только самого медленного, не их суммы. Для IO-bound задач (HTTP, БД, файлы) это даёт огромный прирост.

Event loop - сердце asyncio

asyncio не использует threading - всё работает в одном потоке, через event loop. Loop планирует и выполняет короутины:

1. Корутина начинает работу
2. Когда нужно подождать IO - корутина «уступает» loop'у
3. Loop запускает другую корутину
4. Когда IO готов - loop возвращается к первой корутине
5. Повторяется

Это cooperative multitasking. Корутины кооперируются - явно говорят «я жду, дай другим поработать» через await. В Go переключением занимается рантайм, поэтому горутины пишутся как обычный синхронный код без точек await.

async def и await

Базовый синтаксис:

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)   # уступаем управление на 1с
    print("World")

asyncio.run(hello())
# Hello
# (пауза 1с)
# World

async def создаёт coroutine function. Вызов такой функции возвращает coroutine object, не запускает её сразу:

co = hello()    # ничего не происходит
print(type(co))   # <class 'coroutine'>
asyncio.run(co)   # вот теперь запускается

await приостанавливает текущую корутину до завершения awaitable. Внутри await event loop может выполнять другие задачи. Механика та же, что у yield в генераторах: функция замирает и отдаёт управление наружу.

asyncio.run() - запуск программы

asyncio.run(coro) это удобная точка входа:

async def main():
    print("Start")
    await asyncio.sleep(1)
    print("Done")

asyncio.run(main())

Под капотом создаёт event loop, запускает корутину, закрывает loop. Идиоматично использовать одну функцию main() и asyncio.run(main()) как entry point.

Параллельность через gather

asyncio.gather() запускает корутины параллельно:

async def fetch(url):
    print(f"Fetching {url}")
    await asyncio.sleep(1)   # симулируем IO
    print(f"Done {url}")
    return f"Data from {url}"

async def main():
    results = await asyncio.gather(
        fetch("api/a"),
        fetch("api/b"),
        fetch("api/c"),
    )
    print(results)

asyncio.run(main())
# Fetching api/a
# Fetching api/b
# Fetching api/c
# (пауза 1с - все ждут параллельно)
# Done api/a
# Done api/b
# Done api/c
# ['Data from api/a', 'Data from api/b', 'Data from api/c']

Общее время - не 3 секунды, а 1. Все три корутины запустились, ждут IO одновременно. Когда sleep заканчивается - все продолжают.

Task - запланированная корутина

asyncio.create_task() оборачивает корутину в Task и сразу планирует выполнение:

async def main():
    task1 = asyncio.create_task(fetch("api/a"))
    task2 = asyncio.create_task(fetch("api/b"))

    # Здесь они уже работают параллельно

    result1 = await task1
    result2 = await task2

asyncio.run(main())

Разница с gather:

• gather это shortcut для нескольких параллельных + ожидание всех
• create_task + await даёт больший контроль (можно делать что-то между)

sleep - симуляция асинхронности

asyncio.sleep(seconds) это async-версия time.sleep. Не блокирует event loop - уступает на указанное время:

# ПЛОХО - блокирует event loop
async def bad():
    import time
    time.sleep(1)   # блокирует весь loop, другие task стоят

# ХОРОШО - не блокирует
async def good():
    await asyncio.sleep(1)

Если в async-функции вызвать blocking синхронную операцию, весь event loop встанет. Это типичная ловушка.

Async context managers

async def fetch_with_session():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as response:
            data = await response.text()
            return data

async with это контекст-менеджер с асинхронными __aenter__/__aexit__. Используется для async-ресурсов: сетевых соединений, БД-транзакций, файлов в aiofiles.

Обычный with блокировал бы event loop на open/close - неприемлемо для async кода.

Async iteration

async def fetch_pages(urls):
    for url in urls:
        page = await fetch(url)
        yield page

async def main():
    async for page in fetch_pages(["a", "b", "c"]):
        process(page)

asyncio.run(main())

async for для асинхронных итераторов. Между yield генератора event loop может работать.

Future - результат-в-будущем

asyncio.Future это объект, представляющий результат, который будет доступен. Task это подкласс Future. Базовый Future используется редко напрямую - чаще через Task.

async def set_after(future, delay, value):
    await asyncio.sleep(delay)
    future.set_result(value)

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    asyncio.create_task(set_after(future, 1, "done"))
    result = await future
    print(result)

asyncio.run(main())

Обычно работаешь с Task через create_task, не с Future напрямую.

await чего угодно awaitable

await работает с:

• Корутинами (async def)
• Tasks (asyncio.create_task(...))
• Futures (asyncio.Future и подклассы)
• Любым объектом с __await__ (custom awaitables)

Эти три категории называются awaitables.

sync функция вызывает async - нельзя напрямую

# Sync
def main():
    result = fetch_async("/api")   # это coroutine, не результат
    print(result)   # <coroutine object>

# Чтобы запустить из sync-кода - asyncio.run
def main():
    result = asyncio.run(fetch_async("/api"))
    print(result)

Async можно вызывать из sync через asyncio.run (или loop.run_until_complete). Но это блокирующая операция - не делай так часто.

async функция вызывает sync - просто вызывай

async def main():
    data = sync_function()   # обычный вызов, синхронный
    await async_function()    # async вызов с await

Но если sync_function блокирующая (тяжёлый CPU, blocking IO), она остановит event loop. Используй loop.run_in_executor для тяжёлых синхронных вызовов:

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, heavy_sync_function, arg1, arg2)

Это запускает sync-функцию в thread pool, не блокируя event loop.

Конкурентность vs параллельность

Конкурентность - переключение между задачами в одном потоке (asyncio). Параллельность - одновременное выполнение на нескольких CPU (multiprocessing).

asyncio даёт конкурентность, не параллельность. Для CPU-bound задач он не поможет - наоборот, замедлит из-за overhead. Использовать asyncio только для IO-bound работы.

Где asyncio полезен

Правило: если задача проводит время в wait на IO - asyncio даёт прирост. Если в CPU - не даёт.

Распространённые ошибки

1. Забыть await

async def main():
    fetch("/api")   # ВОЗВРАЩАЕТ COROUTINE, не запускает!

asyncio.run(main())
# RuntimeWarning: coroutine 'fetch' was never awaited

Без await корутина создаётся, но не выполняется. Python предупредит, но баг тонкий.

2. Использовать blocking sync в async

async def main():
    import time
    time.sleep(5)   # БЛОКИРУЕТ весь event loop на 5с

    requests.get("https://...")   # БЛОКИРУЕТ (requests sync библиотека)

Используй async-версии: asyncio.sleep, aiohttp вместо requests.

3. Запуск async из sync без необходимости

def get_data():
    return asyncio.run(fetch("/api"))

# Вызов get_data() из другого async кода:
async def main():
    data = get_data()   # СОЗДАЁТ ВЛОЖЕННЫЙ event loop - ОШИБКА

asyncio.run нельзя вызывать из уже работающего event loop. Если уже async-контекст - используй await напрямую.

4. asyncio.gather без except

async def main():
    results = await asyncio.gather(t1, t2, t3)
    # Если одна упадёт - все остальные тоже отменяются

Поведение по умолчанию: при ошибке одной задачи остальные отменяются. Для индивидуальной обработки - return_exceptions=True:

results = await asyncio.gather(t1, t2, t3, return_exceptions=True)
# results может содержать как значения, так и Exception

Сравнение с Go

В Go конкурентность через горутины - легковесные потоки, управляемые runtime:

go fetch("/api/a")   // запуск горутины
go fetch("/api/b")
go fetch("/api/c")

// Ожидание через каналы или WaitGroup

Различия:

• Go: preemptive scheduling - переключение в любой момент
• Python: cooperative - только на await
• Go: настоящая параллельность на нескольких CPU
• Python: один CPU из-за GIL (с asyncio)

asyncio проще mental model (всё в одном потоке), но менее производительно для смешанных IO + CPU workload.

Мини-задание

1. Базовый async/await:

import asyncio

async def task(name, duration):
    print(f"{name} start")
    await asyncio.sleep(duration)
    print(f"{name} done")
    return f"{name} result"

async def main():
    result = await task("A", 1)
    print(result)

asyncio.run(main())

2. Параллельные задачи через gather:

import asyncio
import time

async def fetch(url, delay):
    print(f"Fetching {url}")
    await asyncio.sleep(delay)
    return f"Data from {url}"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("api/a", 1),
        fetch("api/b", 2),
        fetch("api/c", 1.5),
    )
    elapsed = time.perf_counter() - start
    print(f"Got {len(results)} results in {elapsed:.2f}s")
    # ~2s (max of delays), не 4.5s (sum)

asyncio.run(main())

3. Создание задач:

import asyncio

async def background_task(name):
    for i in range(3):
        print(f"{name}: {i}")
        await asyncio.sleep(0.5)

async def main():
    task1 = asyncio.create_task(background_task("worker-1"))
    task2 = asyncio.create_task(background_task("worker-2"))

    # Делаем что-то ещё пока worker-ы работают
    await asyncio.sleep(0.3)
    print("Main doing other work")

    await task1
    await task2

asyncio.run(main())

Что дальше

Освоили основы asyncio. В следующем уроке - продвинутые паттерны: gather, create_task, Queue, Timeout, отмена и обработка ошибок при concurrent задачах.