asyncio: асинхронный Python без боли

Зачем asyncio?

Python имеет GIL — это означает, что потоки не дают прироста производительности для CPU-bound задач. Но для I/O-bound задач (HTTP-запросы, чтение файлов, БД) — asyncio позволяет «ждать» несколько операций одновременно в одном потоке.

Потоки:    Task1 ─wait─ Task2 ─wait─ Task1 ─wait─ ...  (переключение ОС)
asyncio:   Task1 ─await─ Task2 ─await─ Task1 ─await─... (кооперативная многозадачность)

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

import asyncio

async def fetch_data(url: str) -> str:
    await asyncio.sleep(1)  # симуляция I/O
    return f"data from {url}"

async def main():
    result = await fetch_data("https://example.com")
    print(result)

asyncio.run(main())

async def объявляет корутину. await приостанавливает корутину, пока ожидаемое не завершится, позволяя event loop переключиться на другие задачи.

Параллельный запуск: asyncio.gather

Вместо последовательного await:

async def main():
    # Плохо — последовательно, 3 секунды
    r1 = await fetch("url1")  # 1s
    r2 = await fetch("url2")  # 1s
    r3 = await fetch("url3")  # 1s

    # Хорошо — параллельно, ~1 секунда
    r1, r2, r3 = await asyncio.gather(
        fetch("url1"),
        fetch("url2"),
        fetch("url3"),
    )

Tasks: запускаем и забываем

asyncio.create_task планирует корутину на выполнение немедленно, не ожидая результата:

async def background_job():
    await asyncio.sleep(5)
    print("Фоновая задача завершена")

async def main():
    task = asyncio.create_task(background_job())
    print("Продолжаем работу...")
    await task  # ждём, когда нужно

Таймауты: asyncio.wait_for

async def slow_operation():
    await asyncio.sleep(10)

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
    except asyncio.TimeoutError:
        print("Операция превысила таймаут!")

asyncio.Queue: продюсер–потребитель

async def producer(queue: asyncio.Queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)  # сигнал завершения

async def consumer(queue: asyncio.Queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Обработан: {item}")
        queue.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

Семафоры: ограничиваем параллелизм

async def fetch_with_limit(sem, url):
    async with sem:
        await asyncio.sleep(0.5)  # I/O
        return f"result: {url}"

async def main():
    sem = asyncio.Semaphore(5)  # не более 5 одновременно
    tasks = [fetch_with_limit(sem, f"url{i}") for i in range(20)]
    results = await asyncio.gather(*tasks)

asyncio vs threading vs multiprocessing

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

Блокирующий код в корутине

import time

async def bad():
    time.sleep(1)      # блокирует весь event loop!

async def good():
    await asyncio.sleep(1)  # освобождает event loop

Запуск CPU-bound кода — используй executor

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n):
    return sum(i**2 for i in range(n))

async def main():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy, 10_000_000)

Итог

asyncio идеален для высоконагруженных I/O задач: веб-краулеры, API-агрегаторы, боты. Для CPU-bound работы — multiprocessing. Для легаси-библиотек без async-поддержки — loop.run_in_executor.