Какую проблему решает Thread pool?

Какую проблему решает Thread pool?

Thread pool (пул потоков) — это паттерн управления ресурсами, который решает критическую проблему избыточного создания потоков. Вместо создания нового потока для каждой задачи, используется заранее инициализированный набор потоков, которые переиспользуются для выполнения разных задач.

Основная проблема без Thread pool

Создание потока — дорогостоящая операция:

import threading
import time
from datetime import datetime

# Плохой подход: создание потока для каждой задачи
def bad_approach():
    def task(task_id):
        print(f"Task {task_id} started")
        time.sleep(1)
        print(f"Task {task_id} finished")
    
    start_time = time.time()
    
    # Создаём 1000 потоков
    threads = []
    for i in range(1000):
        t = threading.Thread(target=task, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    elapsed = time.time() - start_time
    print(f"Total time: {elapsed:.2f}s")  # Очень медленно!
    # Потребление памяти: огромное (каждый поток — ~1-8 MB)

Проблемы:

• Перерасход памяти — каждый поток требует памяти для стека (~1-8 MB)
• Замедление операционной системы — планирование тысяч потоков — очень затратно
• Contention — слишком много потоков конкурируют за процессорное время
• Непредсказуемая производительность — система может зависнуть

Решение: Thread pool

from concurrent.futures import ThreadPoolExecutor
import time

# Хороший подход: использование пула потоков
def good_approach():
    def task(task_id):
        print(f"Task {task_id} started")
        time.sleep(1)
        print(f"Task {task_id} finished")
        return f"Result {task_id}"
    
    start_time = time.time()
    
    # Создаём пул из 10 потоков (оптимально для системы)
    with ThreadPoolExecutor(max_workers=10) as executor:
        # Отправляем 1000 задач
        futures = [executor.submit(task, i) for i in range(1000)]
        
        # Получаем результаты по мере готовности
        for future in futures:
            result = future.result()
    
    elapsed = time.time() - start_time
    print(f"Total time: {elapsed:.2f}s")  # Намного быстрее!

Как работает Thread pool

Архитектура:

┌─────────────────────────────────────┐
│     Thread Pool (10 рабочих)        │
├─────────────────────────────────────┤
│                                     │
│  Worker 1 ┐                         │
│  Worker 2 ├─> Очередь задач <─────┤ Task 1
│  Worker 3 ┤   (Queue)              │ Task 2
│  ...      ┤                         │ Task 3
│  Worker 10┘                         │ ...
│                                     │ Task 1000
└─────────────────────────────────────┘

Жизненный цикл:

1. Пул создаёт фиксированное количество потоков при инициализации
2. Каждый поток ждёт задачу из очереди (Queue)
3. Когда задача поступает, свободный поток берёт её и выполняет
4. После завершения задачи, поток возвращается в очередь ожидания
5. Процесс повторяется

Практические примеры

1. Загрузка множества файлов с интернета

from concurrent.futures import ThreadPoolExecutor
import requests

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

urls = [
    'https://example.com/file1.zip',
    'https://example.com/file2.zip',
    # ... 100 URLs
]

with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(download_file, urls)
    for result in results:
        print(result)

2. Обработка запросов к БД

from concurrent.futures import ThreadPoolExecutor
import psycopg2
from psycopg2 import pool

db_pool = psycopg2.pool.SimpleConnectionPool(
    1, 5,  # min, max connections
    host='localhost',
    database='mydb',
    user='user',
    password='password'
)

def query_database(user_id):
    conn = db_pool.getconn()
    try:
        cur = conn.cursor()
        cur.execute('SELECT * FROM users WHERE id = %s', (user_id,))
        return cur.fetchone()
    finally:
        db_pool.putconn(conn)

user_ids = range(1, 101)
with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(query_database, user_ids)
    for result in results:
        print(result)

3. Отправка email-ов асинхронно

from concurrent.futures import ThreadPoolExecutor
import smtplib
from email.mime.text import MIMEText

def send_email(recipient, subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = 'noreply@example.com'
    msg['To'] = recipient
    
    with smtplib.SMTP('localhost', 25) as smtp:
        smtp.send_message(msg)
    
    return f"Email sent to {recipient}"

emails = [
    ('user1@example.com', 'Hello', 'Welcome!'),
    ('user2@example.com', 'Hello', 'Welcome!'),
    # ... 1000 emails
]

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(send_email, *email) for email in emails]
    for future in futures:
        print(future.result())

Оптимальное количество потоков

import os
from concurrent.futures import ThreadPoolExecutor

# Для I/O-bound задач (сеть, файлы)
optimal_workers = (os.cpu_count() or 1) * 2 + 1

# Для CPU-bound задач (вычисления)
optimal_workers = os.cpu_count() or 1

print(f"CPU cores: {os.cpu_count()}")
print(f"Optimal workers for I/O: {optimal_workers}")

Преимущества Thread pool

Когда использовать Thread pool

✓ I/O-bound операции — сетевые запросы, файловые операции, БД запросы ✓ Высоконагруженные приложения — веб-серверы, микросервисы ✓ Асинхронная обработка — фоновые задачи, очереди ✗ CPU-bound операции — для них лучше использовать multiprocessing

Thread pool — критический паттерн для любого серьёзного приложения на Python. Без него невозможно эффективно обрабатывать большое количество I/O операций.