Concurrent.futures.ThreadPoolExecutor: примеры (PYTHON)

Базовые примеры использования ThreadPoolExecutor

Пример с отправкой отдельных задач:

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    time.sleep(1)
    return n * n

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    results = [f.result() for f in futures]

print(results)

[0, 1, 4, 9, 16]

Пример использования map():

with ThreadPoolExecutor() as executor:
    results = list(executor.map(task, [2, 3, 4]))

print(results)

[4, 9, 16]

Пример с обработкой исключений:

def risky_task(x):
    if x == 2:
        raise ValueError('Ошибка при x=2')
    return x * 10

with ThreadPoolExecutor() as executor:
    future = executor.submit(risky_task, 2)
    try:
        result = future.result()
    except ValueError as e:
        print(f'Поймано исключение: {e}')

Поймано исключение: Ошибка при x=2

Альтернативные инструменты параллелизма в Python

ProcessPoolExecutor из того же модуля использует процессы вместо потоков. Подходит для CPU-интенсивных задач, обходя ограничения GIL. Недостаток - большие накладные расходы на межпроцессное взаимодействие.

Модуль multiprocessing предоставляет более низкоуровневый контроль над процессами. Позволяет использовать очереди, каналы и разделяемую память. Выбор этого модуля оправдан при сложной синхронизации.

Асинхронные возможности (asyncio) используют корутины и цикл событий. Эффективны для высоконагруженных сетевых приложений. Недостаток - необходимость переписывания кода с использованием async/await.

Библиотеки для распределенных вычислений (Celery, Dask) подходят для масштабирования задач на несколько машин.

Распространенные ошибки при работе с ThreadPoolExecutor

Блокировка основного потока при вызове result() без проверки готовности:

with ThreadPoolExecutor() as executor:
    future = executor.submit(time.sleep, 5)
    # Основной поток блокируется на 5 секунд
    value = future.result()
    print(value)

Использование потоков для CPU-интенсивных задач. Из-за GIL прирост производительности будет минимальным:

def cpu_intensive(x):
    return sum(i*i for i in range(10**6))  # Не ускорится в потоках

Неявное захватывание глобальных переменных, приводящее к состоянию гонки:

counter = 0
def increment():
    global counter
    counter += 1  # Опасная операция без синхронизации

Забывание вызова shutdown() или использование без контекстного менеджера:

executor = ThreadPoolExecutor()
executor.submit(print, 'Задача')
# Потоки могут оставаться активными

Изменения в ThreadPoolExecutor в новых версиях Python

В Python 3.8 изменено значение по умолчанию для max_workers: теперь используется min(32, os.cpu_count() + 4). Это улучшило производительность на многоядерных системах.

Python 3.9 добавил параметр thread_name_prefix в конструктор, позволяя задавать префикс имен потоков для упрощения отладки.

В Python 3.11 были оптимизированы накладные расходы на создание потоков и улучшена интеграция с модулем asyncio через asyncio.to_thread().

Расширенные сценарии применения ThreadPoolExecutor

Использование с таймаутами и обработкой нескольких фьючерсов:

from concurrent.futures import ThreadPoolExecutor, as_completed
import random
import time

def task(id):
    delay = random.uniform(0.1, 1.0)
    time.sleep(delay)
    return id, delay

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(task, i): i for i in range(10)}
    for future in as_completed(futures, timeout=0.5):
        try:
            result = future.result()
            print(f'Задача {result[0]} выполнена за {result[1]:.2f}с')
        except TimeoutError:
            print('Таймаут ожидания результата')

Задача 3 выполнена за 0.12с
Задача 1 выполнена за 0.23с
...

Интеграция с asyncio для смешанной обработки:

import asyncio
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    # Имитация блокирующей операции
    time.sleep(1)
    return 'Результат из потока'

async def main():
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io)
        print(result)

asyncio.run(main())

Результат из потока

Пользовательский инициализатор потоков:

def init_worker():
    # Инициализация ресурсов для каждого потока
    print(f'Инициализация потока')

with ThreadPoolExecutor(
    max_workers=2,
    initializer=init_worker,
    thread_name_prefix='CustomThread'
) as executor:
    executor.submit(lambda: print('Задача выполняется'))

Инициализация потока
Инициализация потока
Задача выполняется

Реализации пула потоков в других языках

Java: ExecutorService с фиксированным пулом потоков.

ExecutorService executor = Executors.newFixedThreadPool(4);
Future future = executor.submit(() -> 42);
Integer result = future.get();
executor.shutdown();

JavaScript (Node.js): Модуль worker_threads для создания потоков.

const { Worker } = require('worker_threads');
const worker = new Worker('task.js');
worker.on('message', (result) => console.log(result));

Go: Используются горутины и каналы без явного создания пула.

go func() {
    // параллельная задача
}()

C#: ThreadPool.QueueUserWorkItem или Task.Run.

Task.Run(() => Console.WriteLine("Задача в пуле потоков"));

Kotlin: Корутины с диспетчерами.

withContext(Dispatchers.IO) {
    // выполнение в потоке
}