asyncio в Python: как перейти с синхронного кода на асинхронный

Вы написали скрипт, который тянет данные с сотни URL. В синхронном варианте на requests он идёт по списку один за другим: пока ждёт ответ от первого сайта, процессор простаивает. Сто запросов по 300 мс — это 30 секунд, из которых почти всё время программа просто ждёт сеть. asyncio позволяет запустить все сто ожиданий одновременно в одном потоке и уложиться в те же 300–500 мс. Но перенести код механически не выйдет: async def нельзя вызвать как обычную функцию, а один синхронный time.sleep посреди корутины останавливает весь цикл событий. Разберём, где привычные приёмы ломаются и чем платить за переход.

Материал для Python 3.11+. Часть API (asyncio.run, asyncio.TaskGroup) появлялась и менялась между версиями — где это важно, отмечаю отдельно.

Главное отличие от потоков: кооперативность

Синхронный код и потоки вытесняются планировщиком ОС в произвольный момент. asyncio работает иначе: одна корутина держит цикл событий, пока сама не отдаст управление на await. Это кооперативная многозадачность — задачи переключаются только на точках ожидания.

Отсюда следует главное правило перехода: внутри корутины нельзя делать блокирующие вызовы. time.sleep(1), requests.get(), синхронное чтение большого файла — каждый из них останавливает не одну задачу, а весь цикл. Пока такой вызов работает, ни одна другая корутина не выполняется.

Асинхронность не делает код быстрее сама по себе. Она убирает простой на ожидании ввода-вывода. Если ваша задача считает числа в цикле, а не ждёт сеть или диск — asyncio не поможет, нужны процессы.

Три способа получить блокировку и что делать

Перенося синхронный код, вы почти наверняка притащите за собой блокирующие вызовы. Вот как они выглядят и чем их заменить.

1. Синхронная сетевая библиотека

requests, urllib, старые SDK баз данных — синхронные. Внутри корутины они блокируют цикл. Замена — асинхронный аналог: aiohttp или httpx для HTTP, asyncpg для PostgreSQL, aiofiles для файлов.

# requirements: httpx>=0.27
import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> int:
    # await отдаёт управление циклу, пока ждём ответ
    response = await client.get(url, timeout=10)
    return response.status_code

async def main(urls: list[str]) -> None:
    async with httpx.AsyncClient() as client:
        # gather запускает все запросы конкурентно
        results = await asyncio.gather(
            *(fetch(client, url) for url in urls)
        )
    print(results)

if __name__ == "__main__":
    asyncio.run(main(["https://example.com"] * 100))

2. Тяжёлый синхронный вызов, которому нет асинхронной замены

Иногда переписать нельзя: библиотека только синхронная, или это CPU-нагрузка. Тогда вынесите вызов в отдельный поток через asyncio.to_thread (Python 3.9+). Цикл продолжит крутиться, пока поток работает.

import asyncio

def heavy_sync_call(data: bytes) -> bytes:
    # синхронная функция из чужой библиотеки
    return process_image(data)  # условно

async def handle(data: bytes) -> bytes:
    # выполнится в пуле потоков, цикл не блокируется
    return await asyncio.to_thread(heavy_sync_call, data)

Для чистой CPU-нагрузки потоки не спасут из-за GIL — нужен ProcessPoolExecutor через loop.run_in_executor. Это отдельная тема, здесь только обозначаю.

3. time.sleep вместо asyncio.sleep

Классическая ошибка. time.sleep(2) замораживает весь цикл на две секунды. Нужен await asyncio.sleep(2) — он отдаёт управление, и другие задачи работают в это время.

Как запускать корутины: gather, TaskGroup, create_task

Вызов async def возвращает корутину, но не выполняет её. Чтобы код заработал, корутину нужно передать в цикл событий. Способов несколько, и выбор влияет на обработку ошибок.

Для нового кода на 3.11+ предпочтительнее TaskGroup: он не оставляет висящих задач при сбое.

# Python 3.11+
import asyncio

async def worker(number: int) -> int:
    await asyncio.sleep(1)
    if number == 3:
        raise ValueError("сбой в задаче 3")
    return number * 10

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(worker(i)) for i in range(5)]
    except* ValueError as eg:
        # ExceptionGroup: обрабатываем через except*
        print(f"поймали ошибки: {eg.exceptions}")
    else:
        print([t.result() for t in tasks])

asyncio.run(main())

Where это ломается

Перечислю ошибки, на которых спотыкаются те, кто пришёл из синхронного мира.

• Забытый await. Вы вызвали fetch(url) без await, получили объект корутины и пошли дальше. Кода не выполнится, а в логах появится RuntimeWarning: coroutine 'fetch' was never awaited.
• Блокирующий вызов внутри корутины. Признак — один медленный запрос тормозит все остальные, хотя вы ждали конкуренции. Ищите синхронные библиотеки и time.sleep.
• Fire-and-forget без ссылки на задачу. asyncio.create_task(coro()) без сохранения результата: сборщик мусора может удалить задачу до завершения. Держите ссылку в переменной или коллекции.
• Смешивание синхронного и асинхронного клиента БД. Синхронный psycopg внутри asyncio-приложения блокирует цикл на каждом запросе — производительность падает ниже синхронной версии из-за накладных расходов.
• Неограниченный gather на тысячах задач. Тысяча одновременных HTTP-запросов положит и вас, и целевой сервер. Ограничивайте конкурентность через asyncio.Semaphore.

Ограничение конкурентности семафором

import asyncio
import httpx

async def fetch_limited(
    client: httpx.AsyncClient,
    semaphore: asyncio.Semaphore,
    url: str,
) -> int:
    # не больше 10 запросов одновременно
    async with semaphore:
        response = await client.get(url, timeout=10)
        return response.status_code

async def main(urls: list[str]) -> None:
    semaphore = asyncio.Semaphore(10)
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            *(fetch_limited(client, semaphore, url) for url in urls)
        )
    print(results)

Когда переход оправдан, а когда нет

asyncio даёт выигрыш там, где программа много ждёт ввод-вывод и мало считает: сотни сетевых запросов, веб-сервер с медленными бэкендами, работа с несколькими API одновременно. Один-два запроса или скрипт, который считает данные в памяти, переписывать под async смысла нет — сложность вырастет, скорость нет.

Цена перехода реальная: асинхронный код тяжелее отлаживать, стектрейсы длиннее, а одна синхронная библиотека в цепочке обнуляет весь выигрыш. Если экосистема вашей задачи (драйверы БД, SDK) не имеет асинхронных версий — оцените, не проще ли масштабироваться потоками или процессами.

Что делать дальше

Начните с малого: возьмите один скрипт с пачкой сетевых запросов и перепишите его на httpx + asyncio.gather, замерьте время до и после на своём объёме данных. Дальше смотрите в сторону anyio — библиотеки, которая даёт единый интерфейс поверх asyncio и упрощает структурную конкурентность. Для веб-приложений изучите ASGI-фреймворки (FastAPI, Starlette): там asyncio уже встроен в модель обработки запросов. И держите в проде метрику — среднее и p99 времени обработки задачи под реальной параллельностью, а не на одном запросе.