concurrent.futures в Python: распараллелить задачу за 10 минут

У вас есть цикл, который скачивает 200 страниц по очереди, и каждая ждёт ответа сервера по полсекунды. Итого — минута с лишним, из которых процессор простаивает почти всё время. Модуль concurrent.futures из стандартной библиотеки Python позволяет раскидать эти запросы по нескольким потокам и уложиться в единицы секунд. Кода — около десяти строк, устанавливать ничего не надо. Ниже — рабочий пример на Python 3.11+, который вы скопируете и запустите.

Разберём два исполнителя — ThreadPoolExecutor и ProcessPoolExecutor, покажем, когда какой брать, и как не потерять исключения, которые случаются внутри задач. Инвалидация результатов, очереди с приоритетами и async/await — отдельные темы, здесь только базовый параллелизм задач.

Что понадобится

• Python 3.8 или новее. Примеры проверены на 3.11 и 3.12; модуль есть начиная с 3.2, но метод map с параметром chunksize и удобная обработка ошибок стабильны в актуальных версиях.
• Ничего устанавливать не нужно — concurrent.futures входит в стандартную библиотеку.
• Задача, которую можно разбить на независимые части: обработать список файлов, сделать пачку HTTP-запросов, посчитать что-то по каждому элементу коллекции.

Главное различие: потоки или процессы

Выбор исполнителя зависит от того, чем занята ваша задача. В CPython есть GIL — глобальная блокировка интерпретатора^*, из-за которой в один момент времени байт-код Python исполняет только один поток. Поэтому:

• ThreadPoolExecutor — для задач, которые ждут ввода-вывода: сеть, диск, база данных. Пока один поток ждёт ответа, GIL освобождается, и работают остальные.
• ProcessPoolExecutor — для задач, которые грузят процессор: обработка изображений, вычисления, парсинг больших текстов. Каждый процесс — отдельный интерпретатор со своим GIL, поэтому работают настоящие ядра.

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

Самый простой вариант: executor.map

Если у вас есть функция и список аргументов, метод map заменяет обычный цикл почти без изменений. Он возвращает результаты в том же порядке, в каком поданы входные данные.

from concurrent.futures import ThreadPoolExecutor
import urllib.request

def fetch_status(url):
    # Возвращаем HTTP-код ответа для одного адреса
    with urllib.request.urlopen(url, timeout=10) as response:
        return url, response.status

urls = [
    "https://example.com",
    "https://httpbin.org/get",
    "https://www.python.org",
]

# max_workers подбирается опытным путём; для I/O можно ставить больше числа ядер
with ThreadPoolExecutor(max_workers=8) as executor:
    for url, status in executor.map(fetch_status, urls):
        print(f"{status}  {url}")

Что проверить: в консоли появятся строки вида 200 https://example.com для каждого адреса, и все три отработают почти одновременно, а не одна за другой. Блок with сам дождётся завершения всех задач и закроет пул.

Когда нужен контроль: submit и as_completed

У map есть ограничение: он отдаёт результаты строго по порядку и роняет весь итератор на первой же ошибке. Если задачи разной длительности и вы хотите обрабатывать результаты по мере готовности, а сбой одной не должен ронять остальные — используйте submit и as_completed.

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
import urllib.error

def fetch_status(url):
    with urllib.request.urlopen(url, timeout=10) as response:
        return response.status

urls = [
    "https://example.com",
    "https://httpbin.org/status/500",
    "https://nonexistent.invalid",  # намеренно упадёт
]

with ThreadPoolExecutor(max_workers=8) as executor:
    # Запоминаем, какой future какому url соответствует
    future_to_url = {executor.submit(fetch_status, u): u for u in urls}

    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            status = future.result()  # здесь всплывёт исключение задачи, если оно было
            print(f"OK    {status}  {url}")
        except urllib.error.URLError as exc:
            print(f"FAIL  {url}: {exc}")

Что проверить: успешные адреса выведут OK, а несуществующий домен — строку FAIL с текстом ошибки, но программа не упадёт целиком. Порядок вывода будет зависеть от того, что ответит быстрее.

Ключевая идея: future.result() не просто возвращает значение — он повторно возбуждает исключение, которое случилось внутри задачи. Если вы никогда не вызываете result(), ошибка внутри потока проглатывается молча, и вы даже не узнаете, что половина задач упала.

Нагрузка на процессор: меняем один класс

Для вычислительной задачи меняется только имя исполнителя. Функция должна быть определена на верхнем уровне модуля, иначе её не удастся передать в дочерний процесс через pickle.

from concurrent.futures import ProcessPoolExecutor

def heavy_calc(n):
    # Искусственная нагрузка на CPU
    total = 0
    for i in range(n):
        total += i * i
    return total

if __name__ == "__main__":  # обязательно для ProcessPoolExecutor на Windows и macOS
    numbers = [10_000_000] * 8
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(heavy_calc, numbers))
    print(sum(results))

Что проверить: на многоядерной машине эта версия завершится заметно быстрее, чем последовательный цикл. Сравните командой time python script.py для последовательного и параллельного вариантов.

Обёртка if __name__ == "__main__" — не украшение. На macOS (метод старта spawn по умолчанию с 3.8) и на Windows дочерний процесс заново импортирует ваш модуль, и без этой защиты вы получите бесконечное рекурсивное создание процессов.

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

• Для процессов отправная точка — число ядер: os.cpu_count(). Больше смысла обычно нет, процессы начнут бороться за ядра. Это же значение стоит по умолчанию.
• Для потоков при I/O можно ставить десятки — потоки почти всё время ждут, и лишние почти ничего не стоят. По умолчанию с Python 3.8 это min(32, os.cpu_count() + 4). Ориентируйтесь на лимиты внешнего сервиса, а не на свои ядра: 500 одновременных запросов к одному API вас забанят раньше, чем ускорят.

Если не получилось

• Программа зависла или плодит процессы без конца — забыли обернуть код в if __name__ == "__main__" при использовании ProcessPoolExecutor. Добавьте защиту.
• PicklingError: Can't pickle ... — передали в ProcessPoolExecutor лямбду, локальную функцию или объект с несериализуемым состоянием. Вынесите функцию на верхний уровень модуля или переключитесь на потоки, если задача про I/O.
• Результаты приходят, но ошибки пропадают — вы не вызываете future.result(). Пройдитесь по future через as_completed и вызовите result() в блоке try/except.
• BrokenProcessPool — один из дочерних процессов упал жёстко (например, сегфолт в C-расширении или его убил OOM-киллер). Пул после этого непригоден, весь блок нужно перезапускать; проверьте память и стабильность самой функции.
• Параллельная версия работает не быстрее — скорее всего, задача CPU-bound, а вы взяли ThreadPoolExecutor. Из-за GIL потоки не ускорят вычисления. Возьмите процессы.

Что дальше

Замерьте реальный выигрыш на своих данных: time python script.py для последовательной и параллельной версий. Если задача целиком про сеть и вы делаете тысячи запросов — присмотритесь к asyncio с aiohttp: там нет накладных расходов на потоки. Если задача про CPU и данные большие — обратите внимание на параметр chunksize у map для ProcessPoolExecutor: он снижает число операций сериализации. Официальная документация модуля описывает оба исполнителя и все параметры.