⏱️ Раздел 19 · Вопрос #13

Как указать свой Executor для CompletableFuture

Все асинхронные методы класса CompletableFuture имеют перегруженные версии, принимающие в качестве второго параметра пользовательский java.util.concurrent.Executor:

⏱️ CompletableFuture и асинхронность 📄 Исходный Markdown

🟢 Junior Level

Все асинхронные методы класса CompletableFuture имеют перегруженные версии, принимающие в качестве второго параметра пользовательский java.util.concurrent.Executor:

// 1. Создание кастомного пула потоков
ExecutorService customExecutor = Executors.newFixedThreadPool(10);

// 2. Запуск асинхронной операции в кастомном пуле
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("Выполняется в: " + Thread.currentThread().getName());
    return "Результат";
}, customExecutor);

// 3. Продолжение цепочки в кастомном пуле
future.thenApplyAsync(s -> s.toUpperCase(), customExecutor)
      .thenAcceptAsync(System.out::println, customExecutor);

Зачем указывать свой Executor

  1. Изоляция ресурсов (Bulkhead): предотвращение взаимного влияния тяжёлых I/O задач на системный ForkJoinPool.commonPool() и параллельные стримы.
  2. Контроль параллелизма: ограничение максимального числа одновременных подключений к базе данных или внешним API.
  3. Мониторинг и диагностика: присвоение потокам осмысленных имён (http-client-worker-1) для удобного профилирования и анализа thread dump (jstack).

🟡 Middle Level

Главная ловушка: потеря экзекутора в цепочке

В CompletableFuture экземпляр Executor не наследуется автоматически вниз по конвейеру. Если вы передали кастомный пул в supplyAsync(), но забыли передать его в последующий *Async метод, выполнение незаметно “утечёт” в системный ForkJoinPool.commonPool():

// ❌ АНТИПАТТЕРН: первый этап в customExecutor, второй — в общем commonPool!
CompletableFuture.supplyAsync(() -> downloadFile(), ioExecutor)
    .thenApplyAsync(file -> processFile(file)); // УТЕЧКА В commonPool!

// ✅ ПРАВИЛЬНО: передавать executor на каждом асинхронном звене
CompletableFuture.supplyAsync(() -> downloadFile(), ioExecutor)
    .thenApplyAsync(file -> processFile(file), ioExecutor);

Настройка Production-ready пула потоков

В высоконагруженных системах никогда не используют Executors.newCachedThreadPool() (риск бесконечного создания потоков и OOM) или unbounded очереди new LinkedBlockingQueue<>() (риск скрытого накопления задач в памяти):

@Bean(name = "orderProcessingExecutor")
public ExecutorService orderProcessingExecutor() {
    return new ThreadPoolExecutor(
        10,                          // corePoolSize
        50,                          // maximumPoolSize
        60L, TimeUnit.SECONDS,       // keepAliveTime для избыточных потоков
        new ArrayBlockingQueue<>(500),// ОГРАНИЧЕННАЯ ОЧЕРЕДЬ (Bounded queue)
        new ThreadFactoryBuilder()
            .setNameFormat("order-exec-%d")
            .setDaemon(true)
            .build(),
        new ThreadPoolExecutor.CallerRunsPolicy() // Плавное противодавление (Backpressure)
    );
}

[!TIP] Политика CallerRunsPolicy при переполнении очереди заставляет вызывающий поток исполнить задачу самостоятельно. Это естественным образом замедляет входящий поток запросов (Rate Limiting) без падения с ошибкой.

Virtual Threads (Java 21+) для I/O задач

В современных версиях Java для задач ввода-вывода (HTTP-клиенты, JDBC, очереди сообщений) оптимально передавать VirtualThreadPerTaskExecutor:

ExecutorService vtExecutor = Executors.newVirtualThreadPerTaskExecutor();

CompletableFuture.supplyAsync(() -> httpClient.get(url), vtExecutor);

🔴 Senior Level

Переопределение defaultExecutor() (Java 9+)

Начиная с Java 9, в CompletableFuture появился защищённый метод:

public Executor defaultExecutor() {
    return ASYNC_POOL;
}

Это позволяет архитекторам библиотек и фреймворков создавать собственные подклассы CompletableFuture, в которых методы *Async() без параметров будут по умолчанию использовать заданный корпоративный пул вместо ForkJoinPool.commonPool():

public class CustomCompletableFuture<T> extends CompletableFuture<T> {
    private static final ExecutorService DEDICATED_POOL = Executors.newFixedThreadPool(32);

    @Override
    public Executor defaultExecutor() {
        return DEDICATED_POOL; // Подменяем дефолтный пул для всего конвейера!
    }

    @Override
    public <U> CompletableFuture<U> newIncompleteFuture() {
        return new CustomCompletableFuture<>();
    }
}

Формула расчёта размера пула потоков (Брайан Гетц)

Размер пула потоков не выбирается “на глаз”, а рассчитывается исходя из профиля нагрузки:

\[N_{\text{threads}} = N_{\text{cpu}} \times U_{\text{cpu}} \times \left(1 + \frac{W}{C}\right)\]

Где:

  • $N_{\text{cpu}}$: количество доступных ядер процессора.
  • $U_{\text{cpu}}$: целевая утилизация процессора ($0 \le U_{\text{cpu}} \le 1$).
  • $W$: время ожидания ответа (Wait time, блокировка на I/O).
  • $C$: время чистых вычислений (Compute / Service time на CPU).

Пример: На 8 ядрах для микросервиса, где запрос к БД занимает 90 мс ожидания ($W = 90$) и 10 мс обработки на CPU ($C = 10$): \(N = 8 \times 1.0 \times \left(1 + \frac{90}{10}\right) = 8 \times 10 = 80 \text{ потоков}.\)

4 Tricky Questions

1. Принимает ли метод CompletableFuture.allOf() аргумент Executor? Как запустить последующую обработку результатов allOf в кастомном пуле?

Ответ: Нет, метод CompletableFuture.allOf() не имеет перегрузки с параметром Executor! Сигнатура метода: allOf(CompletableFuture<?>... cfs). Метод allOf лишь строит дерево координации завершения. Чтобы обработать результат объединения в кастомном пуле, параметр Executor передаётся в последующий метод конвейера:

CompletableFuture.allOf(f1, f2, f3)
    .thenRunAsync(() -> processAll(), customExecutor); // Выполняется в customExecutor

2. Что произойдёт, если очередь задач кастомного ThreadPoolExecutor переполнена, и сработает стандартная политика AbortPolicy?

Ответ: Пул выбросит RejectedExecutionException:

  • Если отказ произошёл при создании фьючера через CompletableFuture.supplyAsync(supplier, fullExecutor), исключение RejectedExecutionException немедленно выбросится синхронно прямо в вызывающем потоке.
  • Если отказ произошёл на промежуточном асинхронном звене .thenApplyAsync(fn, fullExecutor), то RejectedExecutionException будет перехвачено рантаймом CompletableFuture и завёрнуто в CompletionException, переведя зависимый фьючерс в состояние ошибки.

3. Почему использование Executors.newCachedThreadPool() в связке с CompletableFuture под высокой нагрузкой считается критической уязвимостью?

Ответ: Executors.newCachedThreadPool() имеет corePoolSize = 0, maximumPoolSize = Integer.MAX_VALUE и использует SynchronousQueue. Если на сервис придёт резкий всплеск из 10 000 параллельных асинхронных запросов, пул попытается создать 10 000 системных потоков ОС. Это мгновенно приведёт к исчерпанию оперативной памяти на стеки потоков (-Xss), колоссальным накладным расходам планировщика ядра на переключение контекста и гарантированному краху JVM с java.lang.OutOfMemoryError: unable to create new native thread.

4. Что будет, если передать null в качестве Executor в метод supplyAsync(supplier, null)?

Ответ: Метод немедленно выбросит NullPointerException. Внутри класса CompletableFuture аргумент проверяется через метод screenExecutor(Executor e):

static Executor screenExecutor(Executor e) {
    if (e == null) throw new NullPointerException();
    return e;
}

Значение null не интерпретируется как “использовать пул по умолчанию”; для использования пула по умолчанию следует вызывать одноаргументную перегрузку метода (supplyAsync(supplier)).

🎯 Шпаргалка для интервью

Главное за 60 секунд

  • Для кастомного пула используются перегрузки методов с аргументом Executor (supplyAsync(task, exec), thenApplyAsync(fn, exec) и др.).
  • В CompletableFuture экзекутор не наследуется вниз по цепочке: каждый *Async без экзекутора улетает в commonPool.
  • В production:
    • Bounded очередь (ArrayBlockingQueue).
    • Понятные имена потоков через ThreadFactory.
    • Политика противодавления CallerRunsPolicy.
    • Virtual Threads (newVirtualThreadPerTaskExecutor) в Java 21+ для I/O.
  • Начиная с Java 9, дефолтный пул можно переопределить для своего подкласса через метод defaultExecutor().

Типичные ошибки на интервью

  • Полагать, что достаточно передать Executor только в supplyAsync(), и вся дальнейшая цепочка автоматически останется в этом пуле.
  • Использовать newCachedThreadPool() для высоконагруженных задач.
  • Передавать null в качестве экзекутора в надежде на возврат к поведению по умолчанию (будет выброшен NullPointerException).

Связанные вопросы