Асинхронность в Python: когда она помогает, а когда только мешает

Асинхронность часто воспринимают как способ «сделать быстрее». Это верно ровно для одного класса задач и совершенно неверно для другого — и разница объясняет большинство разочарований.

Два разных «медленно»

Задача, ограниченная вводом-выводом, большую часть времени ждёт: ответа базы, сети, диска. Процессор в это время простаивает.

Задача, ограниченная процессором, всё это время считает: разбирает данные, сжимает изображение, перемножает матрицы.

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

Где выигрыш реален

Классический пример — десятки независимых сетевых запросов. Последовательно это сумма всех задержек, конкурентно — примерно время самого медленного:

import asyncio
import httpx


async def fetch(client, url):
    response = await client.get(url, timeout=10.0)
    return url, response.status_code


async def check_all(urls):
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, url) for url in urls]
        # return_exceptions=True: один упавший запрос не отменяет остальные
        return await asyncio.gather(*tasks, return_exceptions=True)


results = asyncio.run(check_all(urls))

Тридцать запросов по 200 мс — это шесть секунд последовательно и примерно четверть секунды конкурентно.

Одна блокирующая строка портит всё

Главная практическая опасность: событийный цикл однопоточный. Любой синхронный вызов останавливает все корутины, а не только текущую.

async def handler():
    # Останавливает весь цикл на две секунды: ни один запрос не обработается
    time.sleep(2)

    # Правильно
    await asyncio.sleep(2)

То же касается синхронных драйверов базы данных, requests, чтения больших файлов и тяжёлых вычислений. Если избежать блокирующего вызова нельзя, его выносят в отдельный поток:

result = await asyncio.to_thread(blocking_function, arg)

А процессорные задачи — в отдельные процессы, через ProcessPoolExecutor: там GIL уже не мешает.

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

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

semaphore = asyncio.Semaphore(20)


async def fetch_limited(client, url):
    async with semaphore:
        return await fetch(client, url)

Отмена и таймауты

Задача, которая может выполняться неопределённо долго, должна иметь ограничение по времени:

async with asyncio.timeout(5):
    data = await slow_operation()

Отмена в asyncio реализована через исключение CancelledError. Важное правило: его не следует «проглатывать» в except Exception — иначе задача перестанет отменяться. Освобождение ресурсов делают в finally.

Как решить, нужна ли асинхронность

Три вопроса:

1. Программа больше ждёт или больше считает? Считает — асинхронность не поможет, нужны процессы.
2. Много ли одновременных операций ожидания? Один запрос в базу от асинхронности не выиграет ничего.
3. Есть ли асинхронные драйверы для всего стека? Один синхронный драйвер в цепочке сводит выигрыш к нулю.

Что в итоге

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