Какие знаешь правила, для создания пула потоков?

Правила создания пула потоков в Java

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

Выбор типа пула потоков

Java предоставляет различные реализации через ExecutorService:

// 1. FixedThreadPool — фиксированное количество потоков
ExecutorService fixedPool = Executors.newFixedThreadPool(10);
// Хорош для: задач с предсказуемой нагрузкой, CPU-bound операций
// Риск: queue может расти неограниченно и исчерпать память

// 2. CachedThreadPool — динамическое создание потоков
ExecutorService cachedPool = Executors.newCachedThreadPool();
// Хорош для: асинхронных операций, I/O-bound задач
// Риск: может создать слишком много потоков при всплеске нагрузки

// 3. SingleThreadExecutor — один поток
ExecutorService singlePool = Executors.newSingleThreadExecutor();
// Хорош для: последовательного выполнения задач, гарантирует порядок
// Риск: узкое место при высокой нагрузке

// 4. ForkJoinPool — для параллельных алгоритмов (divide-and-conquer)
ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
// Хорош для: рекурсивных задач, работает с work-stealing

// 5. ScheduledExecutorService — для периодических задач
ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(5);
// Хорош для: задач по расписанию, delayed execution

Размер пула потоков

Определение оптимального размера — критическое решение:

Для CPU-bound задач (обработка, вычисления):

// Размер пула ≈ количество доступных процессоров
int poolSize = Runtime.getRuntime().availableProcessors();
// Обычно: 4-16 потоков для современного сервера

Для I/O-bound задач (сетевые запросы, БД):

// poolSize = availableProcessors × (1 + waitTime / computeTime)
// Пример: если задача 80% ждёт I/O и 20% вычисляет
int poolSize = Runtime.getRuntime().availableProcessors() * (1 + 80 / 20);
// ≈ 4 × 5 = 20 потоков

// Или используй практическую рекомендацию
int optimalPoolSize = Runtime.getRuntime().availableProcessors() * 2;

Конфигурация очереди (Queue)

Очень важный параметр, влияет на поведение под нагрузкой:

BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    5,                          // corePoolSize
    15,                         // maximumPoolSize
    60,                         // keepAliveTime
    TimeUnit.SECONDS,
    queue,
    new ThreadFactory() {
        private int count = 0;
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "Worker-" + (++count));
            t.setDaemon(false);
            return t;
        }
    },
    new ThreadPoolExecutor.AbortPolicy() // Policy при отказе
);

Типы очередей:

1. LinkedBlockingQueue (неограниченная) — риск OutOfMemoryError

BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
// Без лимита! Может накопиться много задач

1. ArrayBlockingQueue (ограниченная) — более предсказуемо

BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
// Max 100 задач в очереди, потом rejection policy

1. SynchronousQueue (без буфера) — immediately transfers

BlockingQueue<Runnable> queue = new SynchronousQueue<>();
// Нет очереди, сразу создаёт новый поток или отклоняет

Параметры пула потоков

int corePoolSize = 5;           // Число постоянных потоков
int maximumPoolSize = 20;       // Максимум потоков
long keepAliveTime = 60;        // Время жизни лишних потоков
TimeUnit unit = TimeUnit.SECONDS;
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    corePoolSize,
    maximumPoolSize,
    keepAliveTime,
    unit,
    queue
);

// Настройка поведения
executor.allowCoreThreadTimeOut(true); // Core потоки тоже могут умереть
executor.setRejectedExecutionHandler(
    new ThreadPoolExecutor.CallerRunsPolicy() // Выполнить в основном потоке
);

Rejection Policies — что делать, если пул переполнен?

// 1. AbortPolicy — выбросить исключение (по умолчанию)
executor.setRejectedExecutionHandler(
    new ThreadPoolExecutor.AbortPolicy()
);

// 2. CallerRunsPolicy — выполнить в потоке-отправителе
executor.setRejectedExecutionHandler(
    new ThreadPoolExecutor.CallerRunsPolicy()
);

// 3. DiscardPolicy — молча отбросить задачу
executor.setRejectedExecutionHandler(
    new ThreadPoolExecutor.DiscardPolicy()
);

// 4. DiscardOldestPolicy — отбросить самую старую задачу и вставить новую
executor.setRejectedExecutionHandler(
    new ThreadPoolExecutor.DiscardOldestPolicy()
);

Именование потоков

Исторически потоки имеют худшие имена ("pool-1-thread-2"). Используй ThreadFactory:

ThreadFactory threadFactory = new ThreadFactory() {
    private final AtomicInteger count = new AtomicInteger(0);
    
    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r);
        thread.setName("HttpClient-Worker-" + count.incrementAndGet());
        thread.setDaemon(false);  // Non-daemon для graceful shutdown
        return thread;
    }
};

ExecutorService executor = Executors.newFixedThreadPool(10, threadFactory);

Graceful Shutdown

Правильное завершение пула критично:

// Правильный способ shutdown
executor.shutdown(); // Не принимает новые задачи

try {
    // Ждём, пока текущие задачи завершатся (max 60 сек)
    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
        executor.shutdownNow(); // Force shutdown
        
        // Ждём завершения interrupted задач
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            System.err.println("Pool не завершился");
        }
    }
} catch (InterruptedException ie) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}

Мониторинг пула потоков

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(5);

// Получение статистики
int activeCount = executor.getActiveCount();           // Текущие активные потоки
int coreCount = executor.getCorePoolSize();            // Основные потоки
int maxCount = executor.getMaximumPoolSize();          // Максимум потоков
long completedCount = executor.getCompletedTaskCount(); // Выполнено задач
long totalCount = executor.getTaskCount();             // Всего задач

System.out.printf("Active: %d, Total: %d, Completed: %d\n", 
    activeCount, totalCount, completedCount);

Лучшие практики

1. Никогда не используй Executors.newCachedThreadPool() на production без ограничений
2. Установи лимит на очередь — не более 100-1000 задач
3. Используй ThreadFactory для правильного именования
4. Настрой rejection policy в соответствии с логикой приложения
5. Graceful shutdown при остановке приложения
6. Мониторь метрики пула (active threads, queue size)
7. Тестируй под нагрузкой, не угадывай размер пула
8. Используй ForkJoinPool для рекурсивных задач