Параллелизм и конкурентность в Python: когда и как выбирать

Краткое содержание

• Объясняем разницу между конкурентностью и параллелизмом простыми словами.
• Описываем основные инструменты Python: threading, asyncio, multiprocessing и concurrent.futures.
• Приводим практические советы, когда выбирать каждый инструмент и типовые ошибки.
• Даём контрольный список, методологию выбора и примеры тестов приёмки.

Понимание основных понятий

Определения в одну строку:

• Конкурентность — способность управлять несколькими задачами так, будто они идут одновременно, часто путём переключения контекста.
• Параллелизм — действительно одновременное выполнение нескольких задач, обычно на разных CPU-ядрах.
• GIL — Global Interpreter Lock: ограничение в CPython, которое блокирует одновременное выполнение байткода в нескольких потоках.

Важно: конкурентность и параллелизм не взаимоисключающие: система может быть и конкурентной, и параллельной одновременно.

Почему это важно

• Эффективность использования ресурсов: конкурентность помогает не терять время ожидания ввода/вывода.
• Отзывчивость: интерфейсы и серверы остаются отзывчивыми при высокой латентности I/O.
• Производительность: для тяжёлых вычислений параллелизм на нескольких ядрах заметно ускоряет работу.
• Масштабируемость: выбор модели влияет на архитектуру и операционные расходы.

Ментальные модели и эвристики

• Если код тратит большую часть времени на ожидание сети или диска → сначала думайте о конкурентности (asyncio, потоки, пул потоков).
• Если код активно использует CPU и занимает процессорные циклы → думайте о параллелизме (процессы, пул процессов, распределённые фреймворки).
• Если данные большие и дорого копируются между процессами → оцените стоимость сериализации и передачи данных.

Конкурентность в Python

Конкурентность — естественный выбор для I/O-bound задач: веб-запросы, работа с базами данных, чтение/запись файлов, взаимодействие с внешними сервисами.

Потоки (threading)

Потоки подходят для I/O-bound задач. В CPython потоки ограничены GIL в части выполнения байткода, но при операциях ввода-вывода поток освобождает GIL, поэтому многопоточные I/O часто дают выигрыш.

Пример: многопотечная загрузка URL (упрощённый, без обработки ошибок и таймаутов):

import requests
import time
import threading

urls = [
    'https://www.google.com',
    'https://www.wikipedia.org',
    'https://www.makeuseof.com',
]

def download_url(url):
    response = requests.get(url)
    print(f"Downloaded {url} - Status Code: {response.status_code}")

# Последовательный запуск
start_time = time.time()
for url in urls:
    download_url(url)
end_time = time.time()
print(f"Sequential download took {end_time - start_time:.2f} seconds\n")

# Запуск в потоках
start_time = time.time()
threads = []
for url in urls:
    thread = threading.Thread(target=download_url, args=(url,))
    thread.start()
    threads.append(thread)

for thread in threads:
    thread.join()
end_time = time.time()
print(f"Threaded download took {end_time - start_time:.2f} seconds")

Замечания:

• Добавьте обработку исключений и таймауты (requests.get(…, timeout=…)).
• Потоки совместно используют память процесса — это удобно для общих структур, но требует синхронизации.

Асинхронное программирование с asyncio

asyncio использует цикл событий и корутины для лёгкой конкурентности без потоков. Подходит для большого числа коротких I/O-операций с низкой задержкой.

Пример с aiohttp:

import asyncio
import aiohttp
import time

urls = [
    'https://www.google.com',
    'https://www.wikipedia.org',
    'https://www.makeuseof.com',
]

async def download_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            content = await response.text()
            print(f"Downloaded {url} - Status Code: {response.status}")

async def main():
    tasks = [download_url(url) for url in urls]
    await asyncio.gather(*tasks)

start_time = time.time()
asyncio.run(main())
end_time = time.time()
print(f"Asyncio download took {end_time - start_time:.2f} seconds")

Советы:

• Используйте совместимые с asyncio библиотеки (aiohttp, aioredis и т.д.).
• Для библиотек, не поддерживающих async, применяйте loop.run_in_executor.

concurrent.futures — унифицированный интерфейс

concurrent.futures предоставляет ThreadPoolExecutor и ProcessPoolExecutor в едином API, что упрощает переключение между потоками и процессами.

Краткий пример изменения модели без переписывания логики:

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# Для I/O
with ThreadPoolExecutor(max_workers=10) as ex:
    ex.map(download_url, urls)

# Для CPU
with ProcessPoolExecutor(max_workers=4) as ex:
    ex.map(cpu_bound_function, data_chunks)

Параллелизм в Python

Для CPU-bound задач параллелизм даёт реальный выигрыш: несколько процессов могут выполняться одновременно на разных ядрах, обходя GIL.

multiprocessing

Модуль multiprocessing создаёт процессы с отдельным интерпретатором и памятью.

Пример:

import requests
import multiprocessing
import time

urls = [
    'https://www.google.com',
    'https://www.wikipedia.org',
    'https://www.makeuseof.com',
]

def download_url(url):
    response = requests.get(url)
    print(f"Downloaded {url} - Status Code: {response.status_code}")

def main():
    num_processes = min(len(urls), multiprocessing.cpu_count())
    pool = multiprocessing.Pool(processes=num_processes)

    start_time = time.time()
    pool.map(download_url, urls)
    end_time = time.time()

    pool.close()
    pool.join()

    print(f"Multiprocessing download took {end_time-start_time:.2f} seconds")

if __name__ == '__main__':
    main()

Замечания:

• Передача больших объектов между процессами дорога из-за сериализации.
• Для долгоживущих воркеров используйте менеджеры, очереди или шаремем (shared memory) в Python 3.8+.

Когда использовать конкурентность, а когда параллелизм

• I/O-bound (сеть, диск, БД): конкурентность (asyncio, потоки).
• CPU-bound (числовые расчёты, обработка изображений): параллелизм (процессы, C-расширения, параллельные библиотеки).
• Ограничение памяти и обмен данными: если нужно разделять большой объём данных без копирования, рассмотрите многопоточность или специальные шаримые структуры.

Альтернативные подходы и инструменты

• joblib — удобен для параллельного выполнения вычислений и кэширования результатов (часто применяется в ML).
• Dask — масштабируемые массивы и датафреймы с параллелизмом и распределением задач.
• Ray — распределённые вычисления и actor-модель для масштабируемых приложений.
• C-расширения / numba / cython — убрать GIL для узких участков кода.
• subprocess — запуск внешних программ для изоляции и масштабирования.

Контрпримеры и когда подход не работает

• Малые, короткие задачи: накладные расходы на создание процессов/пулов могут перевесить выгоду.
• Библиотеки, не освобождающие GIL при I/O: многопоточность не даст ускорения.
• Очень большие объёмы данных: сериализация для процессов может стать узким местом.

Методология выбора (мини-процесс)

1. Измерьте: профилируйте и определите, CPU-bound или I/O-bound.
2. Оцените данные: сколько передаётся между задачами; сколько памяти нужно.
3. Оцените устойчивость: нужна ли изоляция задач (процессы) или совместный доступ (потоки).
4. Выберите инструмент: asyncio / ThreadPool / ProcessPool / распределённый фреймворк.
5. Напишите тесты нагрузочного тестирования и SLO/SLA.
6. Мониторьте в продакшене и корректируйте число воркеров.

Роль‑ориентированные чек‑листы

Разработчик:

• Определил тип задачи (I/O/CPU).
• Выбрал совместимые библиотеки (async vs sync).
• Добавил таймауты и обработку исключений.

Код‑ревьюер:

• Проверил отсутствие блокирующих вызовов в async-корутинах.
• Убедился в наличии ограничений конкурентности (семафоры, лимиты).

Оператор/DevOps:

• Настроил метрики (latency, throughput, CPU, memory).
• Прописал политики автоматического масштабирования и restart.

Критерии приёмки

• Функциональность: все URL загружаются успешно в тестовой выборке.
• Производительность: latency и throughput соответствуют ожидаемым SLO.
• Надёжность: система корректно восстанавливается после ошибок ресурсов.
• Ресурсы: использование CPU и памяти в заданных пределах.

Тесты и критерии проверки

• Тест нагрузкой: симулировать N одновременных запросов и измерить P99 latency.
• Тест устойчивости: принудительное падение одного воркера не должно ломать всю обработку.
• Стресс‑тест: увеличение размера данных для оценки сериализации между процессами.

Практические советы и подводные камни

• Не блокируйте event loop: долгое вычисление в корутине заблокирует остальные задачи.
• Используйте bounded семафоры или ограничения очереди, чтобы избежать OOM при всплесках трафика.
• Для смешанных нагрузок комбинируйте asyncio с run_in_executor.
• Для долгих загрузок больших файлов подумайте о стриминге и chunked-обработке.

Безопасность и конфиденциальность

• Всегда валидируйте входные данные при приёме сетевых ответов.
• При параллельной обработке файлов следите за правами доступа и гонками записей.
• Не логируйте чувствительные данные в дебаг-логи, которые могут быть доступны многим процессам.

Матрица сравнения — кратко

• asyncio: лучшая масштабируемость для большого количества коротких I/O операций.
• threading: простота и общая память — когда библиотеки блокируют GIL или работают с C-расширениями.
• multiprocessing: реальный параллелизм для CPU-bound задач, дороже по памяти.
• распределённые фреймворки: масштабирование на кластеры, сложнее в эксплуатации.

Примеры подходящих сценариев

• Веб‑сервер с большим числом одновременных соединений → asyncio или thread pool.
• Пакетная обработка изображений или ML-обучение → multiprocessing / Dask / Ray.
• ETL‑пайплайн с большими объёмами данных → Dask / Ray / распределённые очереди.

Шаблон плана миграции с потоков на asyncio

1. Найти блокирующие места через профилирование.
2. Заменить синхронные HTTP/DB клиенты на async-версии.
3. Вынести тяжёлые синхронные функции в run_in_executor при необходимости.
4. Добавить интеграционные тесты и нагрузочное тестирование.

Короткая памятка (cheat sheet)

• Для I/O → asyncio или ThreadPoolExecutor.
• Для CPU → ProcessPoolExecutor или multiprocessing.
• Для распределения → Dask, Ray, Celery (задачи), Kafka (очереди).

Короткое резюме

• Выбирайте модель по типу нагрузки и стоимости передачи данных.
• Профилируйте перед оптимизацией.
• Тестируйте устойчивость и следите за ресурсами в продакшене.

1‑строчный глоссарий

GIL — Global Interpreter Lock; coroutine — функция, которую можно приостанавливать и возобновлять; I/O-bound — задача, ограниченная вводом/выводом; CPU-bound — задача, ограниченная процессорными циклами.

Заключение

Понимание различий между конкурентностью и параллелизмом и знание инструментов Python помогут выбрать правильную модель выполнения задач. Сначала измеряйте и профилируйте, затем применяйте подходящую стратегию и тестируйте в условиях, близких к продакшену.