Асинхронность в Python: asyncio без магии, на примерах

Асинхронность — главный инструмент Python для сетевых задач: парсинг сотен тысяч страниц, стриминг ответов ChatGPT, чат-серверы на тысячи соединений. В этой статье — asyncio без «магии event loop»: что происходит по шагам, когда он даёт 10-кратный рост, а когда не даёт ничего.

Код: Python 3.12, только стандартная библиотека + aiohttp для сетевых примеров.

Ожидание — вот и вся причина

Возьмём задачу: скачать 20 страниц. Синхронно каждый запрос — это «послал → ждёшь ответ ничего не делая → получил». Если страница отвечает за 0.5 секунды, 20 страниц = 10 секунд чистого ожидания. Процессор при этом занят на 0%.

Asyncio убирает именно это ожидание: пока одна страница грузится, скачиваются остальные девятнадцать. Те же 20 страниц — за ~0.6 секунды. Никакой магии: один процесс, одно ядро, просто ноль простоя.

import asyncio
import time

import aiohttp

URLS = [f"https://example.com/page/{i}" for i in range(20)]


async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as r:
        return await r.text()


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*(fetch(session, u) for u in URLS))
    print(f"Скачано {len(results)} страниц")


if __name__ == "__main__":
    start = time.perf_counter()
    asyncio.run(main())
    print(f"За {time.perf_counter() - start:.1f}с")

Корутина, задача, gather: три слова, которые путают всех

• Корутина — функция с async def. Сама по себе не выполняется: вызов fetch(...) возвращает объект-корутину.
• Задача (task) — корутина, запланированная в event loop. Создаётся через asyncio.create_task() или asyncio.gather(). Именно задачи выполняются конкурентно.
• gather — запускает несколько корутин как задачи и ждёт всех. Результат — список в исходном порядке, независимо от порядка завершения.

Минимальный пример с ручным управлением задачами:

async def brew_coffee() -> str:
    await asyncio.sleep(2)  # имитация ожидания: только asyncio.sleep, не time.sleep!
    return "кофе"

async def toast_bread() -> str:
    await asyncio.sleep(3)
    return "тост"

async def breakfast() -> None:
    task_coffee = asyncio.create_task(brew_coffee())
    task_toast = asyncio.create_task(toast_bread())
    coffee = await task_coffee   # ждём кофе; тост тем временем жарится
    toast = await task_toast
    print(coffee, toast)         # всё готово за 3 секунды, а не за 5

asyncio.run(breakfast())

Ключевая мысль: await — это не «блокировка», а «точка, где можно переключиться на другую задачу». Пока мы ждём task_coffee, event loop честно работает над task_toast.

Семафор: конкурентность с предохранителем

«Запустить всё сразу через gather» — первая ошибка новичка. Тысяча одновременных запросов = тысяча открытых соединений = бан или исчерпание файловых дескрипторов. Семафор ограничивает число одновременных задач:

SEMAPHORE = asyncio.Semaphore(10)  # не более 10 одновременных загрузок

async def fetch_bounded(session: aiohttp.ClientSession, url: str) -> str:
    async with SEMAPHORE:
        return await fetch(session, url)

async def main() -> None:
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*(fetch_bounded(session, u) for u in URLS))

Это тот же приём, что и в статье о парсинге с прокси — только там семафор ограничивал нагрузку на чужой сайт, а здесь — на свои ресурсы.

Четыре ошибки, которые молча убивают async-код

Ошибки asyncio коварны: код работает, результат правильный, а скорость — нет. Вот что проверять в первую очередь.

1. Блокирующий вызов внутри корутины. Самое частое. requests.get(), time.sleep(), синхронный psycopg — любой из них останавливает весь event loop, и все тысяча корутин встают в очередь за ним:

async def broken():
    time.sleep(5)            # БЛОКИРУЕТ всё: никто не работает эти 5 секунд
    requests.get(url)        # то же самое

async def correct():
    await asyncio.sleep(5)   # ждёт, отдав управление другим задачам
    async with aiohttp.ClientSession() as s:
        await s.get(url)

2. Забыли await. fetch(session, url) без await — корутина не выполнилась, вы получили объект и, скорее всего, RuntimeWarning, который заметите через месяц в логах. Включайте строгий режим при отладке: python -W error::RuntimeWarning script.py.

3. Создали задачи, но не дождались. asyncio.create_task() возвращает задачу, которая может не успеть выполниться, если функция завершится раньше. Собирайте ссылки и делайте await на все задачи перед выходом — или используйте gather с самого начала.

4. try/except вокруг gather. Одно исключение в любой корутине роняет весь gather и остальные задачи в лимбе. Для «собрать сколько получится» добавьте return_exceptions=True:

results = await asyncio.gather(*tasks, return_exceptions=True)
ok = [r for r in results if not isinstance(r, Exception)]

Когда async не нужен: честная граница

Asyncio ускоряет ожидание, а не вычисления. Простой тест:

• Скрипт ходит в сеть/БД и 90% времени ждёт → async даст кратный рост.
• Скрипт считает: обрабатывает изображения, гоняет pandas по гигабайту данных, парсит огромный XML → async не поможет вообще. Нужен multiprocessing (процессы в обход GIL) или библиотеки с C-ядром (numpy, lxml).

Правило из практики: профилируйте прежде чем «асинхронить». Часто правильный ответ — не asyncio, а одна очередь с пятью воркерами на Celery.

Итог

Asyncio в трёх предложениях: сетевое ожидание — это потерянное время; корутины переключаются в точках await, чтобы его не терять; семафор и запрет на блокирующие вызовы — цена входа. Освоив это, вы уже сможете читать чужой async-код и писать свой — всё остальное (Queue, wait_for, shield, TaskGroup) — надстройки, которые разбираются за вечер по документации.

Дальше в треке: соединяем asyncio с внешним миром — подключаем ChatGPT к Telegram-боту, где всё ожидание умножается на два API.