Пулы потоков в Java

Создание потока обходится дорого. Каждый new Thread() выделяет около 1 МБ нативного стека, просит ОС запланировать новый ядерный поток и создаёт нагрузку на GC. Программа, выделяющая по одному потоку на задачу, справится с десятью задачами, но сломается при десяти тысячах. Решение — пул потоков: небольшой набор долгоживущих рабочих потоков, которые извлекают задачи из очереди. Пул владеет потоками; вы владеете задачами.

Эта глава посвящена концепции — что такое пул, какие параметры его настраивают и каковы возможные режимы отказа. Следующая глава, Executor framework, знакомит с типами Executor/ExecutorService, с помощью которых вы работаете с пулом. Темы тесно переплетены: эта глава отвечает на вопросы что и зачем, следующая — как.

Зачем нужен пул?

Пул решает три проблемы:

Стоимость создания потока. Выделение нативного стека и запрос нового потока у ОС занимает порядка миллисекунд. Повторное использование существующих потоков — микросекунды. При большой нагрузке это разница между сервером, который выдерживает нагрузку, и тем, который не выдерживает.
Ограничение ресурсов. Платформенный поток на 64-битной JVM занимает около 1 МБ стека — 64 GB ОЗУ хватит примерно на ~64 000 потоков, и у ОС есть собственные накладные расходы на каждый поток. Бесконтрольное создание потоков — это бесконтрольное потребление памяти. Пул ограничивает их количество.
Предсказуемый параллелизм. Пул с N рабочими потоками даёт ровно N параллельных задач. Это значительно лучше подходит для сценария «задействовать все 16 ядер», чем «создавать по потоку на запрос и надеяться».

Цена пулинга: нужно подобрать размер. Слишком маленький — задачи накапливаются в очереди и задержка растёт. Слишком большой — переключение контекста начинает доминировать и пропускная способность падает. Правила подбора размера рассматриваются в главе об executor framework; эта глава посвящена тому, что такое пул.

Анатомия пула

Пул потоков состоит из трёх элементов:

Ограниченный набор рабочих потоков. Рабочие выполняют цикл: берут задачу из очереди, выполняют её, берут следующую и так далее. Они живут всё время существования пула (или до истечения времени простоя — в зависимости от политики).
Очередь задач. Когда вы отправляете работу, а свободных рабочих нет, задача помещается сюда. Тип очереди — LinkedBlockingQueue, ArrayBlockingQueue, SynchronousQueue — полностью определяет поведение пула под нагрузкой.
API отправки задач. execute(Runnable), submit(Callable), invokeAll(...) — способы добавить работу в пул.

В Java всё это реализовано в java.util.concurrent.ThreadPoolExecutor — базовом классе для почти любого пула, с которым вы столкнётесь.

Семь параметров `ThreadPoolExecutor`

Прямая конструкция (которую редко используют напрямую, но именно эти параметры передаёт каждая фабрика):

new ThreadPoolExecutor(
  int corePoolSize,
  int maximumPoolSize,
  long keepAliveTime, TimeUnit unit,
  BlockingQueue<Runnable> workQueue,
  ThreadFactory threadFactory,
  RejectedExecutionHandler handler
);

Параметр	Что управляет
`corePoolSize`	Минимальное количество рабочих, поддерживаемых живыми даже в простое. Потоки до этого числа не уничтожаются.
`maximumPoolSize`	Верхний предел общего числа рабочих. Пул растёт сверх `core` только когда очередь заполнена.
`keepAliveTime`	Сколько времени простаивающий рабочий сверх базового размера ждёт перед завершением.
`workQueue`	Место хранения ожидающих задач. `LinkedBlockingQueue` (неограниченная) vs `ArrayBlockingQueue` (ограниченная) vs `SynchronousQueue` (без буферизации) полностью определяет поведение пула.
`threadFactory`	Способ создания рабочих потоков. Используйте для задания имён, статуса демона, приоритета, обработчиков необработанных исключений.
`handler`	Что происходит, когда заняты и рабочие, и очередь. По умолчанию: `AbortPolicy`.

Неочевидное взаимодействие: пул предпочитает заполнять очередь перед созданием новых потоков сверх core. Поэтому неограниченная очередь означает, что пул никогда не вырастет сверх core — задачи просто накапливаются в очереди бесконечно. Ограниченная очередь (или SynchronousQueue) — это то, что делает параметр max значимым.

Четыре политики отклонения

Когда submit не может принять задачу (очередь заполнена, все рабочие заняты), RejectedExecutionHandler определяет, что произойдёт:

Политика	Поведение
`AbortPolicy` (по умолчанию)	Бросает `RejectedExecutionException`. Вызывающий знает, что задача была отброшена.
`CallerRunsPolicy`	Вызывающий поток сам выполняет задачу. Замедляет вызывающего, обеспечивая обратное давление.
`DiscardPolicy`	Молча отбрасывает задачу. Используйте только для работы типа «лучшее усилие», например телеметрии.
`DiscardOldestPolicy`	Отбрасывает самую старую задачу из очереди и помещает новую. Полезно, когда важна только самая свежая информация.

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

Фабричные методы `Executors` — и почему их стоит избегать

java.util.concurrent.Executors предоставляет удобные фабрики:

Executors.newFixedThreadPool(n);             // core = max = n, unbounded LinkedBlockingQueue
Executors.newCachedThreadPool();             // core = 0, max = Integer.MAX_VALUE, SynchronousQueue, 60s keep-alive
Executors.newSingleThreadExecutor();         // fixed pool with one thread
Executors.newScheduledThreadPool(n);         // for delay/repeat scheduling
Executors.newVirtualThreadPerTaskExecutor(); // Java 21+: one virtual thread per task

У двух из них есть известные ловушки:

newFixedThreadPool использует неограниченную LinkedBlockingQueue. При длительной перегрузке очередь растёт без ограничений — в итоге OOM. Размер пула фиксирован; объём работы, накапливающейся за ним, — нет.
newCachedThreadPool имеет maximum = Integer.MAX_VALUE. При длительном всплеске нагрузки он создаёт потоки без ограничений — в итоге исчерпывает лимит потоков ОС на процесс и крашит JVM.

Они подходят для небольших задач, демонстраций и разовых скриптов. Для production-кода создавайте ThreadPoolExecutor напрямую с ограниченной очередью, разумным max и явной политикой отклонения.

Исключение: newVirtualThreadPerTaskExecutor (Java 21+) раздаёт виртуальные потоки, которые настолько дёшевы, что подход «один на задачу» действительно работает. Это рассматривается в главе о виртуальных потоках.

Жизненный цикл: `shutdown` vs `shutdownNow`

Пул работает до тех пор, пока вы не прикажете ему остановиться. Два режима остановки:

pool.shutdown();                              // stop accepting new work; let queued tasks finish
pool.shutdownNow();                           // stop accepting; interrupt running threads; return queued tasks

boolean terminated = pool.awaitTermination(10, TimeUnit.SECONDS);

shutdown — вежливый вариант: новые отправки не принимаются, текущая работа завершается, затем пул останавливается. shutdownNow — жёсткий вариант: прерываются рабочие, возвращается содержимое очереди. Используйте shutdown для чистого выхода; используйте shutdownNow после shutdown + awaitTermination, если работа не завершилась в срок.

Комбинированный шаблон завершения из документации JDK:

pool.shutdown();
try {
  if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
    pool.shutdownNow();
    pool.awaitTermination(5, TimeUnit.SECONDS);
  }
} catch (InterruptedException e) {
  pool.shutdownNow();
  Thread.currentThread().interrupt();
}

Этот точный шаблон нужен почти всегда в любом коде, владеющем пулом. Без shutdown JVM держит рабочих живыми (они не демоны по умолчанию) и не завершается.

Именование рабочих через `ThreadFactory`

Дефолтная фабрика Executors.defaultThreadFactory() называет потоки pool-1-thread-1, pool-1-thread-2 и т.д. Это лишь немного лучше Thread-7, но всё равно недостаточно. Production-код использует именованную фабрику:

ThreadFactory factory = r -> {
  Thread t = new Thread(r, "image-worker-" + COUNTER.incrementAndGet());
  t.setDaemon(false);
  t.setUncaughtExceptionHandler((thr, ex) -> log.error("uncaught in " + thr.getName(), ex));
  return t;
};
ExecutorService pool = new ThreadPoolExecutor(
    4, 4, 0, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000),
    factory,
    new ThreadPoolExecutor.CallerRunsPolicy());

Фабрика — это ваша возможность задать каждое свойство потока: имя, флаг демона, приоритет, обработчик необработанных исключений, группу потоков. В дампе кучи с 200 потоками поток с именем image-worker-7 — это поток, который можно найти.

Пример: создание ограниченного пула с обратным давлением

Программа ниже создаёт ThreadPoolExecutor с 4 рабочими, ограниченной очередью на 8 элементов и политикой отклонения CallerRunsPolicy — чтобы при перегрузке пула отправитель замедлялся, а не получал исключение.

java— editable, runs on the server

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolDemo {
  public static void main(String[] args) throws InterruptedException {
    AtomicInteger nameCounter = new AtomicInteger();
    ThreadFactory factory = r -> {
      Thread t = new Thread(r, "worker-" + nameCounter.incrementAndGet());
      t.setUncaughtExceptionHandler((thr, ex) ->
          System.out.println("   uncaught on " + thr.getName() + ": " + ex.getMessage()));
      return t;
    };

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        4, 4,                                            // core = max = 4
        0, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(8),                     // bounded queue
        factory,
        new ThreadPoolExecutor.CallerRunsPolicy());      // back-pressure

System.out.println("pool: core=" + pool.getCorePoolSize()
        + "  max=" + pool.getMaximumPoolSize()
        + "  queue capacity=8");

AtomicInteger completed = new AtomicInteger();
    long t0 = System.nanoTime();
    for (int i = 0; i < 40; i++) {
      final int id = i;
      pool.execute(() -> {
        try { Thread.sleep(50); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        completed.incrementAndGet();
        // Print which thread did the work to make back-pressure visible
        String who = Thread.currentThread().getName();
        if (id % 8 == 0) System.out.println("   task " + id + " done on " + who);
      });
    }

// 4 workers x 50ms each + queue of 8 means many tasks will run on `main` via CallerRunsPolicy.
    System.out.println("\nshutting down — waiting up to 5s for all tasks to finish");
    pool.shutdown();
    boolean done = pool.awaitTermination(5, TimeUnit.SECONDS);
    if (!done) {
      pool.shutdownNow();
      pool.awaitTermination(1, TimeUnit.SECONDS);
    }

long ms = (System.nanoTime() - t0) / 1_000_000;
    System.out.println("completed: " + completed.get() + " tasks in " + ms + " ms");
    System.out.println("largest pool size reached: " + pool.getLargestPoolSize());
    System.out.println("task count (recorded): " + pool.getTaskCount());
  }
}

Что стоит вынести из запуска:

Пул имел строгий предел в 4 рабочих потока. При 40 задачах по 50 мс каждая идеализированное последовательное время пула — 40 * 50 / 4 = 500 мс. Реальное время стенных часов было близко к этому — за вычетом задержки из-за CallerRunsPolicy, замедляющей отправителя при заполнении очереди.
Некоторые задачи сообщили имя потока main. Это CallerRunsPolicy в действии: когда очередь заполнена и все рабочие заняты, pool.execute выполняет задачу в вызывающем потоке вместо постановки в очередь или броска исключения. Отправитель замедлился; система осталась ограниченной. Вот как правильно реализуется обратное давление.
pool.getLargestPoolSize() вернул 4 — максимум остался равным базовому размеру. Пул не вырос сверх core даже под устойчивой нагрузкой, потому что ограниченная очередь вмещала кратковременные всплески. При неограниченной очереди (по умолчанию в Executors.newFixedThreadPool) очередь приняла бы все задачи, а largestPoolSize оставался бы равен 4 — но память резко возросла бы, пока задачи накапливались.
Последовательность завершения — это production-шаблон. shutdown() сказал пулу прекратить принимать новые отправки; awaitTermination(5, SECONDS) ждал до 5 секунд завершения выполняемой работы; если работа не завершилась, shutdownNow() прервал бы оставшихся рабочих. Без этих вызовов JVM не завершается — рабочие-не-демоны держат её живой.
Фабрика потоков дала каждому рабочему осмысленное имя (worker-1 ... worker-4) и обработчик необработанных исключений. В production-дампе потоков или профилировщике эти имена — разница между «я знаю, какая это подсистема» и «понятия не имею». Задавайте их для каждого создаваемого пула.

Что дальше

Следующая глава, Java Executor Framework, знакомит с иерархией типов для работы с пулами потоков — Executor, ExecutorService, ScheduledExecutorService — и объясняет, как подобрать размер пула для задач, ограниченных CPU, и для задач, ограниченных вводом-выводом.

Практика

Вы вызываете `Executors.newFixedThreadPool(8)` и отправляете задачи быстрее, чем пул успевает их обрабатывать. В пуле 8 потоков. Каков патологический режим отказа при длительной перегрузке?

Неограниченный рост очереди — `newFixedThreadPool` использует неограниченную `LinkedBlockingQueue`, поэтому память растёт без ограничений и JVM в итоге получает OOMПул молча отбрасывает новые задачи, когда очередь достигает 1024 элементов — лимита по умолчаниюПул бросает `RejectedExecutionException` при каждой отправке, как только выполняются 8 задачПул создаёт новые потоки сверх 8, пока не достигнет лимита потоков ОС