Asyncio в Python: async/await, TaskGroup, потоки, multiprocessing и GIL

Asyncio — встроенная библиотека Python для асинхронного кода. Функции, объявленные через async def, умеют приостанавливаться на await, и пока одна ждёт ответа сети, цикл событий выполняет другие. Так сотни запросов идут одновременно в одном потоке. Но asyncio — лишь один из трёх инструментов: есть ещё потоки (threading) и процессы (multiprocessing). Разберём, как работает asyncio python, что выбрать для вашей задачи и при чём здесь GIL.

Ждать и считать — разные задачи

Без этого различия дальше ничего не понять. Программа тормозит по одной из двух причин:

• I/O-bound — ждёт ввода-вывода: ответа API, базы данных, чтения диска. Процессор при этом простаивает.
• CPU-bound — считает: обрабатывает изображения, гоняет циклы по миллионам записей. Процессор загружен.

Для ожидания подходят asyncio или потоки: пока одна операция ждёт, выполняется другая. Для расчётов нужны настоящие параллельные вычисления на нескольких ядрах — процессы. Перепутать их — главная ошибка: потоки не ускорят расчёт на чистом Python, а процессы ради трёх HTTP-запросов — пушка по воробьям.

Что такое asyncio: корутины и цикл событий

Асинхронный код строится на двух словах. async def объявляет корутину — функцию, которая умеет приостанавливаться. Её вызов не выполняет тело, а создаёт объект корутины. await внутри говорит: «здесь я подожду, а ты пока займись другими задачами». Управляет всем цикл событий (event loop), который запускают через asyncio.run().

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # имитация запроса к API
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5

asyncio.sleep() — асинхронная пауза: она отдаёт управление циклу событий. Обычный time.sleep() внутри корутины заблокирует весь цикл, и остальные задачи будут стоять.

Последовательно или одновременно

Если писать await в цикле, запросы пойдут по очереди — три валюты займут три секунды. Чтобы они шли одновременно, их нужно запустить как задачи.

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # по очереди
    print(f"{time.perf_counter() - start:.1f} с")   # 3.0 с

asyncio.run(main())

TaskGroup: запустить вместе и дождаться всех

Современный способ (Python 3.11+) — группа задач. Внутри блока async with каждая задача стартует сразу через tg.create_task(), а выход из блока происходит, когда закончатся все.

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} с")   # ... 1.0 с

asyncio.run(main())

TaskGroup действует строго: если одна задача упала, остальные отменяются, а все ошибки собираются в ExceptionGroup. Ловят их через except*:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("неизвестная валюта:", eg.exceptions)

asyncio.gather и create_task

До TaskGroup стандартом был asyncio.gather(): он запускает корутины одновременно и возвращает результаты в том же порядке. Его до сих пор много в существующем коде.

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # с return_exceptions=True ошибки вернутся в списке, а не прервут всё

asyncio.create_task() запускает корутину в фоне без группы. Ссылку на задачу нужно сохранить, иначе её может удалить сборщик мусора, а ошибки никто не увидит. Поэтому для новых проектов TaskGroup надёжнее.

Отмена задач и таймауты

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # внутри задачи возникнет CancelledError
    try:
        await task
    except asyncio.CancelledError:
        print("отменена")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11+
            await fetch_rate("EUR")
    except TimeoutError:
        print("слишком долго")

В старых версиях вместо asyncio.timeout() используют asyncio.wait_for(coro, timeout=0.5). Ещё есть asyncio.wait() — он возвращает множества завершённых и незавершённых задач и умеет ждать первую (return_when=asyncio.FIRST_COMPLETED).

В Python 3.15, как показано в курсе, у TaskGroup появился метод tg.cancel(): он отменяет все задачи группы без исключения. Это закрывает сценарий «кто первый»: спросить три сервиса-зеркала, взять первый ответ, остальные отменить. Раньше для этого выбрасывали собственное исключение-сигнал или отменяли задачи вручную.

Синхронизация: Semaphore, Lock, Event, Queue

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

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # не больше 10 одновременно
        await asyncio.sleep(0.1) # здесь был бы запрос
        return url

• asyncio.Lock — только одна корутина в критической секции.
• asyncio.Event — одна корутина ждёт await event.wait(), другая сигналит event.set().
• asyncio.Queue — очередь «производитель — потребитель»: await q.put(), await q.get(), q.task_done().

Грабли: забыли await

Самая частая ошибка — вызвать корутину без await:

async def main():
    rate = fetch_rate("USD")    # забыли await
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited

Python не выполнил функцию, а лишь создал объект корутины. Видите в логах «was never awaited» — ищите пропущенный await. Вторая типичная ошибка — блокирующий вызов (time.sleep, requests.get) внутри корутины. Для HTTP берите асинхронные библиотеки (httpx, aiohttp), а неизбежный синхронный код выносите в поток через await asyncio.to_thread(func, arg).

Многопоточность в Python: модуль threading

Поток (thread) — независимая линия выполнения внутри одного процесса, потоки делят общую память. Для ожидания сети они работают отлично и не требуют переписывать код на async.

import threading, time

def download(name):
    time.sleep(1)               # ожидание сети
    print(name, "готово")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # всё за ~1 секунду

Общая память — и плюс, и источник ошибок: если два потока меняют одну переменную, нужен threading.Lock. Обычно проще не трогать потоки вручную, а взять пул.

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor создаёт пул потоков и раздаёт им задачи. map возвращает результаты в исходном порядке.

from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # всё за ~1 секунду

Что такое GIL и почему потоки не ускоряют расчёты

GIL (Global Interpreter Lock) — глобальная блокировка интерпретатора CPython. В каждый момент байткод Python выполняет только один поток. Пока поток ждёт ввода-вывода, он отпускает GIL, поэтому для сети потоки полезны. А на чистых вычислениях четыре потока работают не быстрее одного: они по очереди передают друг другу блокировку.

from concurrent.futures import ThreadPoolExecutor

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

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # примерно как последовательно

Библиотеки на C (NumPy и подобные) часто отпускают GIL на время тяжёлых операций, поэтому для них потоки ускорение дают.

multiprocessing: настоящая параллельность

У каждого процесса свой интерпретатор и свой GIL, поэтому процессы считают на разных ядрах одновременно. Цена — отдельная память: данные между процессами передаются сериализацией (pickle), и запуск процесса дороже потока.

from concurrent.futures import ProcessPoolExecutor

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

if __name__ == "__main__":      # обязательно на Windows и macOS
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

Без проверки if __name__ == "__main__" дочерние процессы при импорте модуля начнут создавать новые процессы. Есть и низкоуровневый модуль multiprocessing с Process, Pool, Queue, но ProcessPoolExecutor проще и имеет тот же интерфейс, что пул потоков.

Что выбрать: asyncio, threading или multiprocessing

Python без GIL и субинтерпретаторы: что нового

С Python 3.13 существует отдельная сборка без GIL (free-threading, PEP 703) — её помечают буквой t в имени, например python3.15t. В 3.13 она была экспериментальной, с 3.14 (PEP 779) официально поддерживается. В ней потоки считают параллельно на разных ядрах. Но по умолчанию вы по-прежнему получаете обычный Python с GIL: однопоточный код в free-threaded сборке работает немного медленнее, и не все библиотеки к ней готовы.

import sys
print(sys._is_gil_enabled())   # False в free-threaded сборке

Второй путь к параллельным расчётам появился в 3.14 (PEP 734) — субинтерпретаторы в модуле concurrent.interpreters: несколько изолированных интерпретаторов в одном процессе, у каждого свой GIL. Удобнее всего пользоваться ими через InterpreterPoolExecutor, который выглядит так же, как пул потоков. В замере курса четыре тяжёлые задачи заняли 0.39 с против 1.03 с в один поток.

Итог

Asyncio python — инструмент для задач ожидания: корутины на async def, паузы на await, одновременный запуск через TaskGroup или gather. Потоки решают ту же задачу без переписывания кода на async, а для тяжёлых расчётов нужны процессы, потому что GIL не даёт потокам обычного CPython считать параллельно. Free-threading и субинтерпретаторы постепенно меняют эту картину, но базовое правило «ждём — asyncio или потоки, считаем — процессы» пока работает. В курсе всё это разобрано на практике: трекер расходов получает курсы валют параллельно за секунду вместо трёх.