Очереди задач в Java: ExecutorService и ThreadPoolExecutor

Очереди задач в Java: ExecutorService и ThreadPoolExecutor

В многопоточных приложениях на Java управление задачами и их выполнением — одна из ключевых задач. Вместо того чтобы создавать новый поток для каждой задачи (что дорого и неэффективно), разработчики используют очереди задач (Task Queues) в сочетании с пулами потоков. Это позволяет контролировать количество одновременно работающих потоков, переиспользовать их и плавно обрабатывать пиковые нагрузки.

Основой для работы с очередями задач в Java служит фреймворк java.util.concurrent, представленный в Java 5. Главные компоненты: ExecutorService, ThreadPoolExecutor и различные реализации BlockingQueue.

1. ExecutorService — высокоуровневый интерфейс

ExecutorService — это интерфейс, который предоставляет методы для асинхронного выполнения задач. Он скрывает детали управления потоками. Самый популярный способ получить его экземпляр — использовать фабричные методы класса Executors:

ExecutorService executor = Executors.newFixedThreadPool(4); // пул на 4 потока
executor.submit(() -> {
    System.out.println("Задача выполняется в потоке: " + Thread.currentThread().getName());
});
executor.shutdown(); // завершение после выполнения всех задач

Однако фабричные методы имеют ограниченные возможности настройки. Для тонкой настройки используют ThreadPoolExecutor напрямую.

2. ThreadPoolExecutor — настраиваемый пул потоков

ThreadPoolExecutor — это ядро фреймворка. Он принимает очередь задач, политику насыщения, время жизни потоков и другие параметры. Полный конструктор:

public ThreadPoolExecutor(
    int corePoolSize,      // минимальное количество потоков
    int maximumPoolSize,   // максимальное количество потоков
    long keepAliveTime,    // время простоя потока перед удалением
    TimeUnit unit,         // единица времени
    BlockingQueue<Runnable> workQueue, // очередь задач
    RejectedExecutionHandler handler   // политика при переполнении
);

Пример создания пула с пользовательской очередью:

BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    2, 4,
    30, TimeUnit.SECONDS,
    queue,
    new ThreadPoolExecutor.AbortPolicy() // по умолчанию — выброс исключения
);
executor.execute(() -> {
    // тяжелая задача
});
executor.shutdown();

3. BlockingQueue — основа очереди задач

Тип очереди определяет, как задачи будут храниться и передаваться потокам. Наиболее популярные реализации:

• LinkedBlockingQueue — неограниченная очередь на основе связного списка. Если не задать capacity, может привести к переполнению памяти.
• ArrayBlockingQueue — ограниченная очередь на основе массива. Позволяет контролировать количество ожидающих задач.
• SynchronousQueue — очередь нулевой ёмкости. Каждая задача передаётся напрямую потоку, ожидание в очереди отсутствует.
• PriorityBlockingQueue — очередь с приоритетами. Задачи сортируются по приоритету (реализуют Comparable).

Пример с ограниченной очередью и политикой отказа:

BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(10);
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    2, 5,
    10, TimeUnit.SECONDS,
    queue,
    new ThreadPoolExecutor.CallerRunsPolicy() // задача выполняется в вызывающем потоке
);
for (int i = 0; i < 20; i++) {
    executor.submit(() -> {
        Thread.sleep(100);
        return Math.random();
    });
}
executor.shutdown();

4. Политики насыщения (RejectedExecutionHandler)

Когда очередь переполнена и все потоки заняты, ThreadPoolExecutor вызывает обработчик. Стандартные политики:

• AbortPolicy (по умолчанию) — выбрасывает RejectedExecutionException.
• CallerRunsPolicy — задача выполняется в потоке, который вызвал execute().
• DiscardPolicy — молча отбрасывает задачу.
• DiscardOldestPolicy — отбрасывает самую старую невыполненную задачу и повторяет попытку.

Можно создать и собственную политику, реализовав интерфейс RejectedExecutionHandler.

5. CompletableFuture — современный подход

Начиная с Java 8, удобно комбинировать очереди задач с CompletableFuture. Это позволяет строить цепочки асинхронных операций без