⏱️ Розділ 19 · Питання #1

Що таке CompletableFuture і чим він відрізняється від Future

CompletableFuture — це клас, що з'явився в Java 8 у пакеті java.util.concurrent, який реалізує інтерфейси Future та CompletionStage. Він представляє дескриптор результа...

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

🟢 Junior Level

CompletableFuture<T> — це клас, що з’явився в Java 8 у пакеті java.util.concurrent, який реалізує інтерфейси Future<T> та CompletionStage<T>. Він представляє дескриптор результату асинхронного обчислення, який може бути завершений явно (вручну) встановленням значення або винятку.

Головні відмінності від класичного Future

Інтерфейс Future<T>, що існував із Java 5, мав критичний недолік: він був суто пасивним і блокуючим. Щоб отримати результат або зреагувати на завершення завдання, викликаючий потік мусив викликати блокуючий метод get() або опитувати isDone() у циклі.

Характеристика Future<T> (Java 5) CompletableFuture<T> (Java 8+)
Отримання результату Блокуючий get() Неблокуючі колбеки (thenAccept, thenApply)
Композиція завдань Неможлива без блокування потоку Побудова ланцюжків (пайплайнів) будь-якої складності
Ручне завершення Не можна завершити ззовні Метод complete(value) / completeExceptionally(ex)
Обробка помилок try-catch навколо get() Декларативні методи exceptionally(), handle()
Комбінування Немає підтримки Методи allOf(), anyOf(), thenCombine()

Базовий приклад

// ❌ Старий підхід із Future: блокує потік main
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(() -> {
    Thread.sleep(1000);
    return "Дані з БД";
});
String result = future.get(); // ПОТОЧНИЙ ПОТІК ЗАБЛОКОВАНО НА 1 СЕКУНДУ

// ✅ Підхід із CompletableFuture: реактивний і неблокуючий
CompletableFuture.supplyAsync(() -> "Дані з БД")
    .thenApply(data -> data + " -> збагачено")
    .thenAccept(finalResult -> System.out.println("Готово: " + finalResult));
// Потік main не блокується і може одразу виконувати іншу корисну роботу

🟡 Middle Level

Архітектура та ключові інтерфейси

CompletableFuture об’єднує дві ключові концепції:

  1. Future<T>: контракт для перевірки готовності (isDone()), скасування (cancel()) та очікування (get(), join()).
  2. CompletionStage<T>: контракт обіцянки (Promise), що описує етап асинхронного конвеєра. Кожен етап обчислює значення при завершенні іншого етапу без активного очікування.

Життєвий цикл та переведення в завершений стан

На відміну від FutureTask, який переводиться в стан завершення виключно потоком-виконавцем Callable/Runnable, CompletableFuture можна завершити вручну з будь-якого потоку:

CompletableFuture<String> promise = new CompletableFuture<>();

// У фоновому потоці / за надходженням вебсокета / за повідомленням з Kafka:
new Thread(() -> {
    try {
        String data = networkService.fetch();
        promise.complete(data); // Успішне завершення
    } catch (Exception ex) {
        promise.completeExceptionally(ex); // Завершення з помилкою
    }
}).start();

// Клієнтський код підписується на подію без блокування:
promise.thenAccept(System.out::println);

Синхронні та асинхронні методи трансформації

Для кожного типу дії CompletionStage пропонує три варіанти методу:

  • thenApply(Function fn): функція виконується в тому самому потоці, який завершив попередній етап (або у викликаючому потоці, якщо етап уже був завершений).
  • thenApplyAsync(Function fn): функція відправляється на виконання в стандартний пул ForkJoinPool.commonPool().
  • thenApplyAsync(Function fn, Executor executor): функція передається у вказаний користувацький пул потоків.
CompletableFuture.supplyAsync(() -> "User123", customThreadPool)
    // Виконується в потоці з customThreadPool
    .thenApply(id -> id.toLowerCase()) 
    // Виконується у ForkJoinPool.commonPool()
    .thenApplyAsync(id -> repository.findUser(id)) 
    // Виконується у спеціалізованому ioExecutor
    .thenAcceptAsync(user -> notifyService.send(user), ioExecutor);

Типові помилки початківців

  1. Блокування пулу commonPool: запуск довгих мережевих викликів через supplyAsync(() -> httpCall()) без вказівки власного Executor. У стандартному пулі кількість потоків обмежена (Runtime.getRuntime().availableProcessors() - 1), і блокуючий I/O швидко паралізує всі асинхронні завдання в JVM.
  2. Проковтування винятків: якщо до пайплайну не прикріплені exceptionally(), handle() або не викликаний термінальний join(), виняток залишиться непоміченим усередині об’єкта, не потрапивши ні в консоль, ні в логер.

🔴 Senior Level

Внутрішня будова: Lock-free Treiber Stack

Під капотом CompletableFuture реалізований без важких м’ютексів (synchronized або ReentrantLock). Він покладається на атомарні операції CAS (Unsafe / VarHandle залежно від версії Java):

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
    volatile Object result;       // Зберігає або T, або AltResult (виняток/NIL)
    volatile Completion stack;    // Стек залежних етапів (Treiber Stack)
    
    static final class AltResult { 
        final Throwable ex; 
        AltResult(Throwable x) { this.ex = x; } 
    }
    static final AltResult NIL = new AltResult(null); // Маркер для null-результату
}
  1. Поле result:
    • null: обчислення ще не завершено.
    • T: обчислення успішно завершилося об’єктом типу T.
    • NIL: обчислення успішно завершилося зі значенням null.
    • AltResult: обчислення завершилося аварійно (містить Throwable).
  2. Поле stack (Lock-free Treiber Stack):
    • При додаванні підписників (thenApply, thenAccept) створюється вузол Completion.
    • Якщо result == null, новий вузол CAS-операцією проштовхується в голову стека stack.
    • Коли завдання завершується (complete()), потік атомарно через CAS переводить result з null у фінальний стан, витягує стек залежностей та виконує (або шедулить) усі зареєстровані колбеки через метод postComplete().
    • Якщо на момент виклику thenApply поле result вже не null, вузол у стек не кладеться, а дія виконується негайно прямо у викликаючому потоці.

Правило вибору потоку при виконанні синхронних колбеків

Одне з найчастіших джерел перегонів та неочевидних багів — припущення про те, який потік виконує thenApply():

CompletableFuture<String> cf = new CompletableFuture<>();

// Потік А (клієнтський)
cf.thenApply(s -> {
    System.out.println(Thread.currentThread().getName());
    return s.toUpperCase();
});

// Потік Б (воркер)
cf.complete("hello");
  • Сценарій 1: Потік Б викликав complete("hello") ДО того, як Потік А зареєстрував thenApply. У цьому випадку Потік А виконає лямбду у своєму власному потоці синхронно прямо всередині виклику thenApply.
  • Сценарій 2: Потік А зареєстрував thenApply ДО того, як Потік Б завершив ф’ючер. У цьому випадку Потік Б, викликаючи complete(), розпакує стек Completion і виконає лямбду в контексті Потоку Б.
  • Сценарій 3: Викликано thenApplyAsync(...). Лямбда гарантовано відправляється як завдання в пул потоків (воркер пулу).

Обмеження CompletableFuture та відмінність від реактивних потоків

Хоча CompletableFuture реалізує модель Promise/Stage, він розрахований на поодинокі значення (0 або 1 елемент).

  • Він не призначений для обробки потоків даних (data streams) із множинними елементами (на відміну від Flow.Publisher, RxJava або Project Reactor Flux).
  • У ньому відсутня концепція зворотного тиску (Backpressure): виробник не може регулювати темп залежно від можливостей споживача.

4 Tricky Questions

1. У якому потоці виконається виклик .thenApply(fn), якщо батьківський CompletableFuture вже завершений на момент реєстрації колбека?

Відповідь: Колбек виконається синхронно у викликаючому потоці (у тому потоці, який викликає .thenApply()). Усередині CompletableFuture метод перевіряє: якщо result != null, реєстрація в стек залежностей пропускається, і функція fn викликається негайно в поточному потоці. Якщо ж батьківський ф’ючер ще не готовий, колбек покладуть у стек залежностей, і його виконає той потік, який викличе .complete() (або потік завершення асинхронного воркера). Це означає, що поведінка синхронного методу тонко залежить від таймінгів багатопотокового виконання.

2. Чому виклик future.cancel(true) у CompletableFuture не перериває (interrupt) потік, що виконує асинхронне завдання?

Відповідь: У стандартному FutureTask виклик cancel(true) викликає Thread.interrupt() у потоку-виконавця. Проте у CompletableFuture метод cancel(boolean mayInterruptIfRunning) реалізований формально: параметр mayInterruptIfRunning повністю ігнорується! Виклик cancel(true) лише атомарно встановлює стан AltResult(CancellationException) на самому екземплярі CompletableFuture. Потік, який крутить лямбду всередині ForkJoinPool або користувацького Executor, продовжить працювати до природного завершення, якщо тільки завдання саме явно не перевіряє статус скасування через зовнішні прапорці.

3. Що станеться при багаторазовому виклику complete(T) або одночасному виклику complete(val) та completeExceptionally(ex) з різних потоків?

Відповідь: CompletableFuture гарантує потокобезпечну ідемпотентність завершення за рахунок CAS-операції над полем result:

// Перший виклик встановить результат і поверне true:
boolean first = cf.complete("Success"); // true
// Другий виклик нічого не зробить і поверне false:
boolean second = cf.complete("Other");  // false
// Виняток не запишеться:
boolean third = cf.completeExceptionally(new RuntimeException()); // false

Лише перший виклик переведе об’єкт у термінальний стан і запустить залежний стек обчислень. Усі подальші спроби завершити ф’ючер ігноруються, стан об’єкта залишається незмінним.

4. У чому принципова різниця між .get() та .join() з точки зору сигнатури та обробки винятків?

Відповідь:

  • Метод get() оголошений в інтерфейсі Future<T> і кидає перевірювані винятки (checked exceptions): InterruptedException (якщо потік було перервано під час очікування) та ExecutionException (загортає будь-який виняток з асинхронного ланцюжка). Вимагає обов’язкового блоку try-catch.
  • Метод join() оголошений у CompletableFuture і кидає неперевірюваний виняток (unchecked exception) CompletionException, у полі cause якого знаходиться оригінальна помилка (а при скасуванні кидає CancellationException). Він спеціально спроєктований для зручного використання всередині лямбда-виразів та Stream API, де checked-винятки заборонені сигнатурами функціональних інтерфейсів.

🎯 Шпаргалка для інтерв’ю

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

  • CompletableFuture — це реалізація Future + CompletionStage, що з’явилася в Java 8 для неблокуючого асинхронного програмування.
  • Дозволяє пов’язувати завдання в ланцюжки (thenApply, thenCompose), комбінувати результати (thenCombine, allOf) та декларативно обробляти помилки (exceptionally, handle).
  • За замовчуванням асинхронні методи без передачі екзекутора використовують ForkJoinPool.commonPool(). Для будь-яких блокуючих I/O операцій обов’язково передавати виділений ExecutorService.
  • Внутрішня структура: lock-free стек залежностей (Treiber Stack) та volatile Object result, що оновлюються через CAS.

Типові помилки на інтерв’ю

  • Стверджувати, що thenApply() завжди виконується в окремому фоновому потоці. (Ні! Без суфікса *Async він виконується або в потоці завершувача, або у викликаючому потоці).
  • Вважати, що cf.cancel(true) перериває потік через Thread.interrupt(). (Параметр ігнорується, завдання в пулі не переривається).
  • Використовувати CompletableFuture для безперервних потокових даних (для цього потрібен Reactive Streams / RxJava / Project Reactor).

Пов’язані теми