Пул потоков в Python: как работать с ThreadPoolExecutor для параллельных задач

import concurrent.futures
import time

def task(name):
print(f"Задача {name} начата")
time.sleep(2) # Имитация работы
print(f"Задача {name} завершена")
return f"Результат {name}"

# Создаем пул с 3 рабочими потоками
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
# Запускаем 5 задач
future_results = {executor.submit(task, i): i for i in range(5)}

# Обрабатываем результаты по мере завершения
for future in concurrent.futures.as_completed(future_results):
task_id = future_results[future]
try:
result = future.result()
print(f"Задача {task_id} вернула: {result}")
except Exception as e:
print(f"Задача {task_id} вызвала исключение: {e}")

import concurrent.futures
import requests

urls = [
'https://www.python.org/',
'https://docs.python.org/',
'https://pypi.org/',
'https://github.com/python/cpython',
'https://www.djangoproject.com/'
]

def load_url(url):
response = requests.get(url)
return f"{url}: {len(response.content)} bytes"

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
results = executor.map(load_url, urls)

for url, result in zip(urls, results):
print(result)

executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
# ... выполнение задач ...
# Завершить пул без ожидания задач
executor.shutdown(wait=False)

import concurrent.futures
import time
import random

def task(id):
sleep_time = random.uniform(0.5, 3.0)
print(f"Задача {id} начата, будет выполняться {sleep_time:.2f} сек")
time.sleep(sleep_time)
if random.random() < 0.2: # 20% шанс ошибки
raise ValueError(f"Ошибка в задаче {id}")
return f"Результат задачи {id}"

def task_done_callback(future):
# Эта функция вызывается, когда задача завершается
try:
result = future.result()
print(f"Callback: задача завершена успешно с результатом: {result}")
except Exception as e:
print(f"Callback: задача завершилась с ошибкой: {e}")

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = []
for i in range(10):
future = executor.submit(task, i)
future.add_done_callback(task_done_callback)
futures.append(future)

# Дополнительно можем дождаться всех результатов
for i, future in enumerate(futures):
try:
result = future.result()
print(f"Основной поток: задача {i} вернула {result}")
except Exception as e:
print(f"Основной поток: задача {i} вызвала исключение {e}")

with concurrent.futures.ThreadPoolExecutor() as executor:
future_to_url = {executor.submit(load_url, url): url for url in urls}

for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
print(f"{url} обработан, получено {len(data)} байт")
except Exception as e:
print(f"{url} вызвал ошибку: {e}")

future = executor.submit(long_running_task)
try:
result = future.result(timeout=10) # Ждем результат максимум 10 секунд
except concurrent.futures.TimeoutError:
print("Задача выполняется слишком долго, отмена...")
future.cancel() # Пытаемся отменить задачу

done, not_done = concurrent.futures.wait(
futures, 
timeout=5,
return_when=concurrent.futures.FIRST_COMPLETED # Или ALL_COMPLETED, FIRST_EXCEPTION
)

print(f"Завершено {len(done)} задач, осталось {len(not_done)}")

import os
import psutil
import math

# Подход 1: Базовый подход — CPU + 4 (как в стандартной библиотеке)
max_workers_1 = (os.cpu_count() or 1) + 4

# Подход 2: Для I/O-bound задач — количество ядер умножить на 2-3
max_workers_2 = (os.cpu_count() or 1) * 3

# Подход 3: Более сложный подход, учитывающий память
# Предполагаем, что каждый поток использует примерно 10MB памяти
memory_per_thread = 10 * 1024 * 1024 # 10MB в байтах
available_memory = psutil.virtual_memory().available
memory_based_max_workers = int(available_memory * 0.8 / memory_per_thread)

# Выбираем минимальное значение, чтобы не перегрузить систему
final_max_workers = min(max_workers_2, memory_based_max_workers)
print(f"Оптимальное количество потоков: {final_max_workers}")

# Неэффективный подход: отдельный поток для каждой строки файла
def process_file_inefficient(filename):
with concurrent.futures.ThreadPoolExecutor() as executor:
with open(filename) as file:
futures = [executor.submit(process_line, line) for line in file]

for future in concurrent.futures.as_completed(futures):
result = future.result()
# обработка результата

# Оптимизированный подход: разделение файла на блоки
def process_file_efficient(filename, num_workers=None):
if num_workers is None:
num_workers = (os.cpu_count() or 1) * 2

# Определяем размер файла и делим на блоки
file_size = os.path.getsize(filename)
chunk_size = math.ceil(file_size / num_workers)

# Функция обработки блока строк
def process_chunk(start, size):
results = []
with open(filename) as file:
file.seek(start)
bytes_read = 0
for line in file:
results.append(process_line(line))
bytes_read += len(line.encode('utf-8'))
if bytes_read >= size:
break
return results

# Запускаем обработку блоков параллельно
with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
futures = []
for i in range(0, file_size, chunk_size):
futures.append(executor.submit(process_chunk, i, chunk_size))

# Собираем результаты
all_results = []
for future in concurrent.futures.as_completed(futures):
all_results.extend(future.result())

return all_results

import concurrent.futures
import threading
import time
import requests

# Ограничиваем до 5 одновременных запросов
semaphore = threading.Semaphore(5)

def fetch_with_limit(url):
with semaphore: # Блокирует, если уже 5 активных запросов
print(f"Fetching {url}")
response = requests.get(url)
time.sleep(1) # Имитация обработки
return response.status_code

urls = [f"https://example.com/{i}" for i in range(20)]

with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
future_to_url = {executor.submit(fetch_with_limit, url): url for url in urls}

for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
status_code = future.result()
print(f"{url} – Status code: {status_code}")
except Exception as exc:
print(f"{url} generated an exception: {exc}")

import time
import concurrent.futures

def benchmark(func, *args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"{func.__name__} занял {end_time – start_time:.4f} секунд")
return result, end_time – start_time

# Последовательная версия
def process_sequentially(data):
results = []
for item in data:
results.append(process_item(item))
return results

# Параллельная версия с map
def process_with_map(data, max_workers=None):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(process_item, data))
return results

# Параллельная версия с submit и as_completed
def process_with_submit(data, max_workers=None):
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [executor.submit(process_item, item) for item in data]
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
return results

# Сравнение производительности
data = list(range(100))
seq_result, seq_time = benchmark(process_sequentially, data)
map_result, map_time = benchmark(process_with_map, data)
submit_result, submit_time = benchmark(process_with_submit, data)

print(f"Ускорение с map: {seq_time/map_time:.2f}x")
print(f"Ускорение с submit: {seq_time/submit_time:.2f}x")

import concurrent.futures
import queue
import threading
import time
import random

# Очередь задач
task_queue = queue.Queue()
# Флаг для остановки обработчиков
stop_processing = threading.Event()

def worker_task():
while not stop_processing.is_set():
try:
# Получаем задачу из очереди с таймаутом
task_id, task_func, args = task_queue.get(timeout=0.5)
print(f"Обработка задачи {task_id}")
result = task_func(*args)
print(f"Задача {task_id} завершена с результатом: {result}")
except queue.Empty:
# Если очередь пуста, продолжаем цикл
continue
except Exception as e:
print(f"Ошибка при обработке задачи: {e}")

def example_task(a, b):
time.sleep(random.uniform(0.1, 0.5)) # Имитация работы
return a + b

# Создаем и запускаем пул потоков
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
# Запускаем воркеры
workers = [executor.submit(worker_task) for _ in range(3)]

# Генерируем и добавляем задачи в очередь
for i in range(10):
task_queue.put((i, example_task, (i, i*2)))
time.sleep(random.uniform(0.1, 0.3)) # Случайные интервалы между задачами

# Даем время на обработку
time.sleep(5)

# Сигнализируем воркерам о завершении
stop_processing.set()

# Дожидаемся завершения всех воркеров
for worker in workers:
worker.result() # Этот вызов блокирует до завершения потока

import concurrent.futures
import time

def stage1(data):
print(f"Обработка данных {data} на этапе 1")
time.sleep(1) # Имитация сложной обработки
return data * 2

def stage2(data):
print(f"Обработка данных {data} на этапе 2")
time.sleep(0.5)
return data + 10

def stage3(data):
print(f"Обработка данных {data} на этапе 3")
time.sleep(0.7)
return data ** 2

def process_pipeline(input_data):
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
# Запускаем первый этап для всех входных данных
stage1_futures = {
executor.submit(stage1, data): data for data in input_data
}

# Запускаем второй этап по мере завершения первого
stage2_futures = {}
for future in concurrent.futures.as_completed(stage1_futures):
data = stage1_futures[future]
try:
result = future.result()
# Запускаем второй этап с результатом первого
stage2_future = executor.submit(stage2, result)
stage2_futures[stage2_future] = result
except Exception as e:
print(f"Ошибка на этапе 1 для данных {data}: {e}")

# Запускаем третий этап по мере завершения второго
stage3_futures = {}
for future in concurrent.futures.as_completed(stage2_futures):
data = stage2_futures[future]
try:
result = future.result()
# Запускаем третий этап
stage3_future = executor.submit(stage3, result)
stage3_futures[stage3_future] = result
except Exception as e:
print(f"Ошибка на этапе 2 для данных {data}: {e}")

# Собираем финальные результаты
final_results = {}
for future in concurrent.futures.as_completed(stage3_futures):
data = stage3_futures[future]
try:
result = future.result()
final_results[data] = result
except Exception as e:
print(f"Ошибка на этапе 3 для данных {data}: {e}")

return final_results

# Обрабатываем набор данных через конвейер
input_data = [1, 2, 3, 4, 5]
results = process_pipeline(input_data)
for data, result in results.items():
print(f"Входные данные: {data} -> Финальный результат: {result}")

import concurrent.futures
import time
import random

def unreliable_task(task_id):
# Имитация ненадежной задачи, иногда вызывающей ошибки
if random.random() < 0.7: # 70% шанс ошибки
print(f"Задача {task_id}: ошибка!")
raise ValueError(f"Случайная ошибка в задаче {task_id}")
time.sleep(random.uniform(0.1, 0.3))
print(f"Задача {task_id}: успех!")
return f"Результат задачи {task_id}"

def execute_with_retry(task_func, task_id, max_retries=3, backoff_factor=2):
retries = 0
while retries <= max_retries:
try:
return task_func(task_id)
except Exception as e:
wait_time = backoff_factor ** retries
retries += 1
if retries > max_retries:
print(f"Задача {task_id} не удалась после {max_retries} попыток")
raise
print(f"Задача {task_id}: повтор через {wait_time} сек...")
time.sleep(wait_time)

# Используем с ThreadPoolExecutor
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = [
executor.submit(execute_with_retry, unreliable_task, i) 
for i in range(5)
]

for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
print(f"Получен результат: {result}")
except Exception as e:
print(f"Окончательная ошибка: {e}")

import concurrent.futures
import threading
import time
import random

class LeakyBucket:
"""Реализация алгоритма Leaky Bucket для ограничения скорости"""

def __init__(self, capacity, leak_rate):
self.capacity = capacity # Максимальное число токенов
self.leak_rate = leak_rate # Токенов в секунду
self.tokens = capacity # Текущее число токенов
self.lock = threading.Lock() # Для потокобезопасности
self.last_leak = time.time() # Время последней утечки

def consume(self, tokens=1):
"""Потребляет токены, возвращает True если успешно, иначе False"""
with self.lock:
# Обновляем число токенов
now = time.time()
elapsed = now – self.last_leak
leaked = elapsed * self.leak_rate
self.tokens = min(self.capacity, self.tokens + leaked)
self.last_leak = now

# Пытаемся потребить токены
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False

def try_consume(self, tokens=1, max_wait=None):
"""Пытается потребить токены, ожидая при необходимости"""
start_time = time.time()
while True:
if self.consume(tokens):
return True

# Проверяем, не превышено ли максимальное время ожидания
if max_wait is not None and time.time() – start_time > max_wait:
return False

# Спим небольшое время перед повторной попыткой
time.sleep(0.1)

# Создаем ограничитель скорости: 5 запросов в секунду
rate_limiter = LeakyBucket(capacity=5, leak_rate=5)

def rate_limited_task(task_id):
# Пытаемся получить разрешение на выполнение
if not rate_limiter.try_consume(1, max_wait=10):
raise TimeoutError(f"Задача {task_id}: превышено время ожидания в очереди")

print(f"Задача {task_id} начата в {time.strftime('%H:%M:%S')}")
time.sleep(random.uniform(0.1, 0.5)) # Имитация работы
return f"Результат задачи {task_id}"

# Запускаем множество задач, но с ограничением скорости
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = [
executor.submit(rate_limited_task, i) 
for i in range(20)
]

for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
print(result)
except Exception as e:
print(f"Ошибка: {e}")

import concurrent.futures
import threading
import time
import os
import psutil

class DynamicThreadPoolExecutor:
"""Исполнитель пула потоков с динамическим изменением размера"""

def __init__(self, min_workers=1, max_workers=None, 
adjust_interval=60, load_threshold=0.7):
self.min_workers = min_workers
self.max_workers = max_workers or (os.cpu_count() or 1) * 5
self.adjust_interval = adjust_interval # Интервал корректировки в секундах
self.load_threshold = load_threshold # Порог загрузки для увеличения

self.executor = concurrent.futures.ThreadPoolExecutor(
max_workers=self.min_workers
)
self.current_workers = self.min_workers
self.task_count = 0
self.task_lock = threading.Lock()

self.monitor_thread = threading.Thread(
target=self._monitor, daemon=True
)
self.running = True
self.monitor_thread.start()

def _monitor(self):
"""Мониторит нагрузку и корректирует размер пула"""
while self.running:
time.sleep(self.adjust_interval)

# Определяем текущую загрузку пула
with self.task_lock:
if self.task_count > 0:
load = self.task_count / self.current_workers
else:
load = 0

if load > self.load_threshold and self.current_workers < self.max_workers:
# Увеличиваем пул, если загрузка высока
new_size = min(
self.current_workers * 2, 
self.max_workers
)
print(f"Увеличиваем пул потоков: {self.current_workers} -> {new_size}")

# Создаем новый executor с увеличенным размером
new_executor = concurrent.futures.ThreadPoolExecutor(max_workers=new_size)
old_executor = self.executor
self.executor = new_executor
self.current_workers = new_size

# Завершаем старый executor, но позволяем завершить текущие задачи
old_executor.shutdown(wait=False)

elif load < self.load_threshold * 0.3 and self.current_workers > self.min_workers:
# Уменьшаем пул, если загрузка низкая
new_size = max(
self.current_workers // 2, 
self.min_workers
)
print(f"Уменьшаем пул потоков: {self.current_workers} -> {new_size}")

new_executor = concurrent.futures.ThreadPoolExecutor(max_workers=new_size)
old_executor = self.executor
self.executor = new_executor
self.current_workers = new_size

old_executor.shutdown(wait=False)

def submit(self, fn, *args, **kwargs):
"""Отправляет задачу на выполнение"""
with self.task_lock:
self.task_count += 1

future = self.executor.submit(fn, *args, **kwargs)
future.add_done_callback(self._task_done)
return future

def _task_done(self, future):
"""Вызывается при завершении задачи"""
with self.task_lock:
self.task_count -= 1

def shutdown(self, wait=True):
"""Завершает работу пула потоков"""
self.running = False
self.monitor_thread.join(timeout=1)
self.executor.shutdown(wait=wait)

def __enter__(self):
return self

def __exit__(self, exc_type, exc_val, exc_tb):
self.shutdown()

# Пример использования
def example_task(task_id):
print(f"Задача {task_id} выполняется")
time.sleep(random.uniform(1, 3))
return f"Результат {task_id}"

with DynamicThreadPoolExecutor(min_workers=2, max_workers=10) as executor:
# Отправляем волны задач
for wave in range(3):
print(f"\nЗапуск волны {wave} задач")
futures = []

# Создаем много задач в каждой волне
task_count = 5 if wave == 0 else (15 if wave == 1 else 3)
for i in range(task_count):
futures.append(executor.submit(example_task, f"{wave}-{i}"))

# Ждем завершения всех задач в волне
for future in concurrent.futures.as_completed(futures):
result = future.result()
print(f"Получен результат: {result}")

# Пауза между волнами
time.sleep(5)