GIL: почему threading в Python особенный

Global Interpreter Lock — это мьютекс, который не даёт двум потокам Python одновременно исполнять байткод. Следствие: threading не даёт прироста на CPU-задачах. Два потока на двухъядерной машине с CPU-нагрузкой могут работать медленнее, чем один.

Но для I/O-задач (сетевые запросы, чтение файлов, работа с БД) GIL не мешает — пока поток ждёт ответа, он отпускает GIL, и другой поток работает. Поэтому threading для I/O работает, но asyncio делает это лучше и с меньшими накладными расходами.

Три подхода и когда их использовать

Задача               Инструмент         Почему
────────────────────────────────────────────────────
HTTP-запросы         asyncio            1 поток, 1000+ соединений
Чтение файлов        asyncio/threading  зависит от объёма
Работа с БД          asyncio            asyncpg, aiosqlite
CPU: парсинг         multiprocessing    обход GIL
CPU: ML/числа        multiprocessing    обход GIL
Стороннее SDK        threading          если нет async версии
Фоновая задача       threading          простой случай

asyncio: базовый паттерн

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    response = await client.get(url, timeout=10.0)
    response.raise_for_status()
    return {"url": url, "status": response.status_code, "size": len(response.content)}

async def main():
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/uuid",
    ]

    async with httpx.AsyncClient() as client:
        # Запустить все запросы одновременно
        results = await asyncio.gather(*[fetch(client, url) for url in urls])

    for r in results:
        print(f"{r['url']}: {r['status']} ({r['size']} bytes)")

asyncio.run(main())

asyncio.gather запускает все корутины конкурентно — не параллельно, но одновременно. Пока одна ждёт ответа, другие продолжают работу. Суммарное время = max(отдельных времён), а не sum(отдельных времён).

asyncio.gather vs asyncio.create_task

import asyncio

async def task(name: str, delay: float):
    print(f"{name}: начало")
    await asyncio.sleep(delay)
    print(f"{name}: конец")
    return name

async def with_gather():
    # gather ждёт все задачи сразу
    results = await asyncio.gather(
        task("A", 1.0),
        task("B", 0.5),
        task("C", 2.0),
    )
    print(f"Все выполнены: {results}")

async def with_tasks():
    # create_task позволяет задачам работать в фоне
    t1 = asyncio.create_task(task("A", 1.0))
    t2 = asyncio.create_task(task("B", 0.5))

    # Делаем что-то другое пока задачи работают
    print("Работаем пока задачи в фоне...")
    await asyncio.sleep(0.2)
    print("Продолжаем другую работу...")

    # Ждём результатов когда нужно
    a_result = await t1
    b_result = await t2
    return a_result, b_result

asyncio.run(with_gather())

Обработка ошибок в gather

async def risky_task(n: int):
    await asyncio.sleep(0.1)
    if n == 2:
        raise ValueError(f"Ошибка в задаче {n}")
    return n * 10

async def main():
    # return_exceptions=True — ошибки возвращаются как значения, не прерывают остальные
    results = await asyncio.gather(
        risky_task(1),
        risky_task(2),
        risky_task(3),
        return_exceptions=True
    )

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Задача {i+1} упала: {result}")
        else:
            print(f"Задача {i+1}: {result}")

asyncio.run(main())
# Задача 1: 10
# Задача 2 упала: Ошибка в задаче 2
# Задача 3: 30

threading: когда оправдан

Threading нужен когда библиотека не поддерживает async, но нужна одновременная работа нескольких операций: 

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def sync_operation(n: int) -> int:
    """Синхронная операция, например вызов legacy SDK."""
    time.sleep(0.5)  # имитация I/O
    return n * n

def run_parallel(items: list[int], workers: int = 5) -> list[int]:
    results = {}
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(sync_operation, item): item for item in items}
        for future in as_completed(futures):
            item = futures[future]
            try:
                results[item] = future.result()
            except Exception as e:
                print(f"Ошибка для {item}: {e}")
    return [results[i] for i in items if i in results]

print(run_parallel(list(range(10))))

Запуск синхронного кода в asyncio

Если в async-приложении нужно вызвать блокирующую функцию без остановки event loop: 

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def call_blocking_function():
    loop = asyncio.get_event_loop()
    # Запустить sync_operation в отдельном потоке, не блокируя event loop
    result = await loop.run_in_executor(executor, sync_operation, 42)
    return result

# В Python 3.9+ проще:
async def call_blocking_v2():
    result = await asyncio.to_thread(sync_operation, 42)
    return result

multiprocessing: обход GIL для CPU

from multiprocessing import Pool
import numpy as np

def cpu_heavy(data: list) -> float:
    """Тяжёлые вычисления — сюда идут в процессы."""
    arr = np.array(data)
    return float(np.std(arr) * np.mean(arr ** 2))

def process_chunks(chunks: list[list]) -> list[float]:
    with Pool(processes=4) as pool:
        results = pool.map(cpu_heavy, chunks)
    return results

# Данные разбиваем на куски
data = list(range(1_000_000))
chunk_size = len(data) // 4
chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]

results = process_chunks(chunks)
print(results)

Практическое правило выбора

Простой тест: если ваша функция большую часть времени ждёт (сеть, диск, БД) — asyncio. Если большую часть времени считает — multiprocessing. Если это legacy код без async — ThreadPoolExecutor как обёртка.