Асинхронность с asyncio: корутины, await, gather

Разберись в кооперативной многозадачности asyncio: ленивые корутины, точки приостановки, gather, TaskGroup и ловушки блокирующих вызовов.

10 мин · 3 вопросов

asyncio реализует кооперативную многозадачность в одном потоке: цикл событий управляет корутинами, которые добровольно отдают управление на каждом await. Ключевой момент, который часто понимают неверно: вызов функции async def ничего не выполняет. Он создаёт инертный объект-корутину, тело которой начинает двигаться только тогда, когда её ожидают через await или ставят на цикл событий как задачу.

import asyncio

async def greet() -> str:
    return "hello"

coro = greet()
print(type(coro).__name__)

async def main() -> None:
    print(await coro)

asyncio.run(main())

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

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1.0),
        fetch("b", 1.0),
        fetch("c", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(results)
    print(f"total: {elapsed:.1f} s")

asyncio.run(main())

По умолчанию gather пробрасывает первое возникшее исключение, но оставляет остальные задачи работать в фоне - редко именно то, чего ты хочешь. С return_exceptions=True исключения возвращаются как обычные результаты, чтобы разобрать их потом. Начиная с Python 3.11 выбирай asyncio.TaskGroup: при первой же ошибке все соседние задачи корректно отменяются, а несколько исключений поднимаются наверх собранными в ExceptionGroup, который ловится через except*.

import asyncio
import time

def blocking_io() -> str:
    time.sleep(1)
    return "report written"

async def main() -> None:
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1, result="tick"),
    )
    print(results)

asyncio.run(main())

Проверка знаний

Убедись, что запомнил ключевые моменты этого урока.

1. Что даёт вызов fetch('a', 1.0), записанный сам по себе, без await и без create_task?
   • Инертный объект-корутину: ни одна строка тела функции не выполняется
   • Немедленное синхронное выполнение корутины
   • Задачу, автоматически поставленную в цикл событий
   • SyntaxError, потому что await обязателен
2. Три asyncio.sleep(1), запущенные вместе через asyncio.gather: какова общая длительность?
   • Около 3 секунд, ведь код выполняет один-единственный поток
   • Предсказать невозможно, не зная числа ядер
   • Около 1 секунды: ожидания накладываются друг на друга, цикл переключается с одной задачи на другую
3. Почему time.sleep(2) под запретом внутри корутины?
   • Он вызывает исключение в асинхронном контексте
   • Он блокирует поток цикла событий: ни одна другая задача не продвинется в течение 2 секунд
   • Он просто менее точен, чем asyncio.sleep
   • Он потребляет вдвое больше памяти