Як вказати свій Executor для CompletableFuture

Усі асинхронні методи класу CompletableFuture мають перевантажені версії, які приймають як другий параметр користувацький java.util.concurrent.Executor:

🟢 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`

Ізоляція ресурсів (Bulkhead): запобігання взаємному впливу важких I/O завдань на системний ForkJoinPool.commonPool() та паралельні стріми.
Контроль паралелізму: обмеження максимальної кількості одночасних підключень до бази даних або зовнішніх API.
Моніторинг та діагностика: присвоєння потокам зрозумілих імен (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).