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

Краткие примеры использования

Базовый пример с одной функцией:

from concurrent.futures import ThreadPoolExecutor
import time

def task(name, delay):
    time.sleep(delay)
    return f"Задача {name} выполнена"

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(task, "A", 1)
    print(future.result())

Задача A выполнена

Пример с несколькими задачами:

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = []
    for i in range(3):
        future = executor.submit(task, f"Task-{i}", i*0.5)
        futures.append(future)
    
    for f in futures:
        print(f.result())

Задача Task-0 выполнена
Задача Task-1 выполнена
Задача Task-2 выполнена

Пример с передачей именованных аргументов:

def process_data(id, multiplier=1, prefix=""):
    return f"{prefix}ID:{id}, результат:{id*multiplier}"

with ThreadPoolExecutor() as executor:
    f1 = executor.submit(process_data, 5, multiplier=2, prefix="Результат: ")
    f2 = executor.submit(process_data, 10, prefix="Данные: ")
    print(f1.result())
    print(f2.result())

Результат: ID:5, результат:10
Данные: ID:10, результат:10

Похожие функции в Python

ThreadPoolExecutor.map(): применяет функцию к каждому элементу итерируемого объекта параллельно. Возвращает итератор по результатам в порядке отправки задач. Удобен для обработки коллекций данных.

ProcessPoolExecutor.submit(): аналогичен ThreadPoolExecutor.submit, но использует процессы вместо потоков. Предпочтителен для CPU-bound задач (интенсивные вычисления).

asyncio.create_task() и корутины: используются для асинхронного выполнения в одном потоке с кооперативной многозадачностью. Эффективны для I/O-bound операций в асинхронных приложениях.

multiprocessing.pool.Pool.apply_async(): функция из модуля multiprocessing для асинхронного выполнения в пуле процессов. Более низкоуровневый аналог ProcessPoolExecutor.

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

1. Блокировка основного потока при вызове result() до завершения всех задач:

with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(time.sleep, 2) for _ in range(5)]
    # Эта строка заблокируется на 2 секунды для КАЖДОГО future
    for f in futures:
        print(f.result())  # Будет последовательное ожидание

2. Игнорирование исключений в потоках:

def faulty_task():
    raise ValueError("Ошибка в потоке")

with ThreadPoolExecutor() as executor:
    future = executor.submit(faulty_task)
    # Исключение не будет видно, пока не вызвать result() или exception()
    time.sleep(0.5)
    print("Программа продолжается...")
    # Только здесь увидим исключение
    try:
        result = future.result()
    except ValueError as e:
        print(f"Поймано исключение: {e}")

Программа продолжается...
Поймано исключение: Ошибка в потоке

3. Использование несериализуемых или непотокобезопасных объектов:

shared_list = []  # Непотокобезопасный объект

def add_item(item):
    shared_list.append(item)  # Возможна гонка данных

with ThreadPoolExecutor(max_workers=5) as executor:
    for i in range(1000):
        executor.submit(add_item, i)
    
print(f"Длина списка: {len(shared_list)}")  # Может быть меньше 1000

Изменения в последних версиях Python

В Python 3.9 была добавлена поддержка параметра timeout в методах shutdown() и контекстный менеджер теперь ждет завершения всех задач при выходе.

В Python 3.8 появилась возможность использовать ThreadPoolExecutor с контекстным менеджером для автоматического вызова shutdown(wait=True).

Начиная с Python 3.7, объекты Future имеют метод cancel(), который пытается отменить выполнение задачи, если она еще не началась.

В Python 3.5 были улучшены механизмы обработки исключений в потоках - теперь информация о трассировке стека сохраняется лучше.

Расширенные примеры использования

Обработка результатов по мере готовности с помощью as_completed:

from concurrent.futures import ThreadPoolExecutor, as_completed
import random

def simulate_work(task_id):
    delay = random.uniform(0.1, 1.0)
    time.sleep(delay)
    return task_id, delay

with ThreadPoolExecutor(max_workers=3) as executor:
    tasks = {executor.submit(simulate_work, i): i for i in range(10)}
    
    for future in as_completed(tasks):
        task_id = tasks[future]
        try:
            result_id, delay = future.result()
            print(f"Задача {task_id} завершена за {delay:.2f} сек")
        except Exception as e:
            print(f"Задача {task_id} вызвала исключение: {e}")

Задача 2 завершена за 0.12 сек
Задача 0 завершена за 0.23 сек
Задача 1 завершена за 0.34 сек
...

Использование callback-функций для обработки результатов:

def process_result(future):
    try:
        result = future.result()
        print(f"Обработан результат: {result}")
    except Exception as e:
        print(f"Ошибка при обработке: {e}")

def long_calculation(x):
    time.sleep(0.5)
    if x == 3:
        raise ValueError("Особый случай")
    return x * x

with ThreadPoolExecutor() as executor:
    for i in range(5):
        future = executor.submit(long_calculation, i)
        future.add_done_callback(process_result)

Обработан результат: 0
Обработан результат: 1
Обработан результат: 4
Ошибка при обработке: Особый случай
Обработан результат: 16

Комбинирование с ThreadLocal для хранения состояния потока:

import threading

thread_local = threading.local()

def get_thread_data():
    if not hasattr(thread_local, 'counter'):
        thread_local.counter = 0
    thread_local.counter += 1
    return f"Поток {threading.current_thread().name}, счетчик: {thread_local.counter}"

with ThreadPoolExecutor(max_workers=2, thread_name_prefix='Worker') as executor:
    futures = [executor.submit(get_thread_data) for _ in range(6)]
    for f in futures:
        print(f.result())

Поток Worker_0, счетчик: 1
Поток Worker_1, счетчик: 1
Поток Worker_0, счетчик: 2
Поток Worker_1, счетчик: 2
Поток Worker_0, счетчик: 3
Поток Worker_1, счетчик: 3

Ограничение времени выполнения задач с timeout:

def slow_function(seconds):
    time.sleep(seconds)
    return f"Выполнено за {seconds} сек"

with ThreadPoolExecutor() as executor:
    future = executor.submit(slow_function, 5)
    try:
        result = future.result(timeout=2)  # Ждем только 2 секунды
        print(result)
    except TimeoutError:
        print("Превышено время ожидания")
        future.cancel()  # Пытаемся отменить задачу

Превышено время ожидания

Аналоги в других языках программирования

JavaScript (Node.js): Promises и async/await. Отличие - однопоточная модель с event loop.

// Пример с Promise.all()
async function task(name, delay) {
    await new Promise(res => setTimeout(res, delay));
    return `Задача ${name} выполнена`;
}

async function main() {
    const promises = [
        task("A", 1000),
        task("B", 500)
    ];
    const results = await Promise.all(promises);
    console.log(results);
}
main();

[ 'Задача A выполнена', 'Задача B выполнена' ]

Java: ExecutorService.submit(). Возвращает Future, похожий на Python, но с более строгой типизацией.

import java.util.concurrent.*;

ExecutorService executor = Executors.newFixedThreadPool(2);
Future future = executor.submit(() -> {
    Thread.sleep(1000);
    return "Результат";
});
System.out.println(future.get());
executor.shutdown();

Результат

Golang: горутины (goroutines) и каналы (channels). Более легковесные чем потоки.

package main
import (
    "fmt"
    "time"
)

func task(name string, ch chan<- string) {
    time.Sleep(1 * time.Second)
    ch <- fmt.Sprintf("Задача %s выполнена", name)
}

func main() {
    ch := make(chan string, 2)
    go task("A", ch)
    go task("B", ch)
    fmt.Println(<-ch)
    fmt.Println(<-ch)
}

Задача A выполнена
Задача B выполнена