Многопоточная загрузка в Python: рабочие приёмы для I/O

Скрипт качает 500 картинок с CDN по одной ссылке за раз. Каждый запрос — 200–400 мс сетевой задержки, и большую часть времени процесс просто ждёт ответа. На последовательном коде это 3–4 минуты чистого ожидания. Тот же список через пул из 20 потоков закрывается за 15–20 секунд, потому что пока одни потоки ждут сеть, другие уже принимают данные. Разбираем, как это устроить правильно и где приёмы перестают работать.

Речь про I/O-задачи: загрузку по HTTP, чтение с диска, обращения к API. Для счётных задач (парсинг, сжатие, обработка изображений в чистом Python) потоки не помогут из-за GIL — там нужен ProcessPoolExecutor, и это отдельная тема, здесь её не разворачиваем.

Почему очевидное решение медленное

Наивный цикл for url in urls: download(url) тратит время не на процессор, а на ожидание. Пока сокет ждёт ответа сервера, интерпретатор ничего полезного не делает. GIL — глобальная блокировка интерпретатора* — здесь не мешает: во время блокирующего сетевого вызова поток освобождает GIL, и другой поток успевает поработать. Именно поэтому многопоточность в Python хорошо ускоряет сеть и диск, хотя и бесполезна для чистых вычислений.

Примеры проверены на Python 3.11 и 3.12, библиотека requests 2.31.

Подход 1: ThreadPoolExecutor — рабочая лошадка

Не создавайте потоки вручную через threading.Thread для пачки задач. Ручное управление означает, что вы сами следите за завершением, собираете результаты и ловите исключения. concurrent.futures.ThreadPoolExecutor делает это за вас и ограничивает число одновременных потоков.

import concurrent.futures
import requests

def download(url: str) -> tuple[str, int]:
    # Отдельная сессия на поток безопаснее, чем общая
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return url, len(resp.content)

urls = [f"https://httpbin.org/bytes/{n}" for n in range(1000, 1050)]

results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
    futures = {pool.submit(download, u): u for u in urls}
    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        try:
            u, size = future.result()
            results.append((u, size))
        except Exception as exc:
            print(f"Ошибка на {url}: {exc}")

print(f"Готово: {len(results)} из {len(urls)}")

Ключевой момент — as_completed: результаты приходят по мере готовности, а не в порядке подачи. Исключение из download не роняет весь пул, а всплывает при вызове future.result(). Это то место, где новички теряют ошибки: если не вызвать result(), упавшая задача пройдёт молча.

Что выигрываете: простой синхронный код, привычный requests, никакого переписывания на async. Чем платите: каждый поток съедает память под стек (порядка мегабайта), и на тысячах одновременных потоков это заметно.

Сколько ставить max_workers

Для I/O число потоков не привязано к числу ядер — процессор простаивает. Разумный старт — 10–50. Больше 100–200 потоков на процесс упираются в память и в лимиты ОС на открытые сокеты. Точное число подбирают замером под конкретный сервер: если удвоение потоков перестало ускорять загрузку — упёрлись либо в канал, либо в rate-limit на стороне сервера.

Многопоточность не делает сеть быстрее. Она перестаёт ждать по одному и начинает ждать параллельно — а канал и терпение сервера остаются прежними.

Подход 2: переиспользование соединений

Каждый requests.get к одному хосту заново устанавливает TCP- и TLS-соединение. На HTTPS это добавляет один-два round trip к каждому запросу. При загрузке сотен файлов с одного домена это ощутимо. Решение — requests.Session с настроенным пулом соединений.

import requests
from requests.adapters import HTTPAdapter

def make_session(pool_size: int = 20) -> requests.Session:
    session = requests.Session()
    adapter = HTTPAdapter(
        pool_connections=pool_size,
        pool_maxsize=pool_size,
        max_retries=3,
    )
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

Тонкость: объект Session в целом потокобезопасен для одновременных запросов, но общий пул соединений становится узким местом, если pool_maxsize меньше числа потоков — лишние потоки будут ждать освобождения соединения либо получать предупреждение о переполнении пула. Держите pool_maxsize не меньше max_workers.

Подход 3: asyncio, когда потоков становится много

Когда задач тысячи и одновременных соединений нужны сотни, потоки упираются в память. Асинхронный код на asyncio с aiohttp держит те же сотни соединений в одном потоке за счёт кооперативной многозадачности. Цена — вся цепочка вызовов должна быть асинхронной: синхронный requests внутри корутины заблокирует весь цикл событий.

import asyncio
import aiohttp

async def download(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as resp:
        resp.raise_for_status()
        data = await resp.read()
        return url, len(data)

async def main(urls: list[str], limit: int = 50):
    # Семафор ограничивает число одновременных запросов
    sem = asyncio.Semaphore(limit)
    connector = aiohttp.TCPConnector(limit=limit)

    async def bounded(session, url):
        async with sem:
            return await download(session, url)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [bounded(session, u) for u in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# результаты, где ошибки — это объекты исключений
# results = asyncio.run(main(my_urls))

Проверено на aiohttp 3.9 и Python 3.11+. return_exceptions=True в gather не даёт одной упавшей задаче отменить остальные — иначе первое же исключение остановит всю пачку. Ошибки придётся отфильтровать из списка результатов по типу.

Когда что выбирать

Правило простое: до пары сотен одновременных загрузок берите ThreadPoolExecutor — меньше кода и меньше способов ошибиться. Когда упираетесь в память от потоков или нужны тысячи соединений — переходите на asyncio. Переписывать рабочий пул на async «на всякий случай» смысла нет.

Что ломается на практике

• Rate limit сервера. 20 потоков против API, разрешающего 5 запросов в секунду, дадут пачку ответов 429 Too Many Requests. Ограничивайте частоту, а не только параллельность: семафор ограничивает одновременность, но не число запросов в секунду.
• Потерянные исключения. Если не вызвать future.result() или не обработать элементы из return_exceptions=True, упавшие загрузки исчезнут без следа, а в результатах окажется меньше файлов, чем ждали.
• Отсутствие таймаута. Один зависший сервер без timeout заблокирует поток навсегда, и пул постепенно встанет. Таймаут обязателен на каждом запросе.
• Запись в общий список без блокировки. results.append(...) из потоков в целом переживает GIL для операции append, но составные операции (проверить-и-вставить) уже требуют threading.Lock. Безопаснее собирать результаты через as_completed в главном потоке, как в примере выше.
• Исчерпание дескрипторов. Тысячи одновременных соединений упираются в лимит открытых файлов ОС. На Linux смотрите ulimit -n.

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

Начните с ThreadPoolExecutor и общей Session с настроенным пулом — это закрывает большинство задач загрузки. Замерьте время на реальном списке ссылок при 10, 20 и 50 потоках: точка, где рост числа потоков перестаёт ускорять, и есть ваш потолок для этого сервера. Добавьте таймауты и повторы (max_retries в адаптере). Если после этого не хватает — измерьте пиковую память процесса под нагрузкой, и только тогда решайте про переход на asyncio. Для ограничения частоты запросов посмотрите готовые лимитеры вроде токен-бакета, чтобы не собирать 429 руками.

* GIL (Global Interpreter Lock) — блокировка, позволяющая исполнять байт-код Python только одному потоку в каждый момент. На блокирующих I/O-вызовах поток её отпускает, поэтому многопоточность ускоряет сеть и диск, но не счётные задачи.