Як комбінувати результати декількох CompletableFuture

У реальних додатках бізнес-логіка часто вимагає одночасного виконання кількох незалежних операцій: паралельні запити до мікросервісів, читання з кешу та бази даних, звернення до...

🟢 Junior Level

У CompletableFuture передбачено три базові патерни комбінування:

1. Попарне об’єднання двох незалежних завдань (`thenCombine`)

Коли два завдання виконуються паралельно, і після готовності обох потрібно об’єднати їхні результати в один об’єкт:

CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userService.getUser(userId));
CompletableFuture<List<Order>> ordersFuture = CompletableFuture.supplyAsync(() -> orderService.getOrders(userId));

// Виконуються паралельно. Час очікування = max(час_юзера, час_замовлень)
CompletableFuture<UserProfile> profileFuture = userFuture.thenCombine(
    ordersFuture,
    (user, orders) -> new UserProfile(user, orders)
);

2. Очікування завершення множини ($N$) завдань (`allOf`)

Коли потрібно запустити довільну кількість паралельних операцій і дочекатися закінчення всіх:

CompletableFuture<Void> allDone = CompletableFuture.allOf(cf1, cf2, cf3, cf4);

allDone.thenRun(() -> {
    // У цей момент УСІ 4 завдання гарантовано завершені
    System.out.println("Усі запити виконано!");
});

3. Перегони до першої відповіді (`anyOf`)

Коли достатньо отримати результат першого завершеного завдання (наприклад, опитування реплік або кешу):

// Повертає результат найшвидшого джерела
CompletableFuture<Object> fastest = CompletableFuture.anyOf(cacheFuture, replica1Future, replica2Future);
fastest.thenAccept(data -> System.out.println("Отримано першу відповідь: " + data));

🟡 Middle Level

Систематизація методів комбінації

Інтерфейс CompletionStage ділить комбінації на дві групи: логічне І (Both) та логічне АБО (Either).

Тип відношення	Метод трансформації	Метод споживання	Метод побічного ефекту
Обидва завершилися (AND)	`thenCombine(other, BiFunction)`	`thenAcceptBoth(other, BiConsumer)`	`runAfterBoth(other, Runnable)`
Перший завершився (OR)	`applyToEither(other, Function)`	`acceptEither(other, Consumer)`	`runAfterEither(other, Runnable)`
Усі $N$ завершилися (AND)	`CompletableFuture.allOf(...)`	—	—
Будь-який з $N$ завершився (OR)	`CompletableFuture.anyOf(...)`	—	—

Ідіоматичний патерн збору результатів з `allOf` у `List<T>`

Метод CompletableFuture.allOf() повертає CompletableFuture<Void>. Щоб зібрати типізований список CompletableFuture<List<T>> без блокування викликаючого потоку:

public <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> futures) {
    CompletableFuture<Void> allDone = CompletableFuture.allOf(
        futures.toArray(new CompletableFuture[0])
    );

    return allDone.thenApply(v -> 
        futures.stream()
            // join() тут абсолютно безпечний і НЕ блокує потік, 
            // оскільки allOf гарантує, що всі futures уже завершені!
            .map(CompletableFuture::join)
            .toList()
    );
}

Ізоляція помилок при комбінуванні через `allOf`

Якщо хоча б один ф’ючер в allOf падає з помилкою, результуючий ф’ючер також переходить у стан винятку. Щоб одиничний збій не ламав увесь збір даних, помилки локалізують на рівні кожного підзапиту:

CompletableFuture<Weather> weatherFuture = weatherService.getWeatherAsync()
    .exceptionally(ex -> Weather.UNKNOWN); // Локальний fallback

CompletableFuture<Traffic> trafficFuture = trafficService.getTrafficAsync()
    .exceptionally(ex -> Traffic.NORMAL); // Локальний fallback

// Тепер allOf гарантовано не впаде через один збій:
CompletableFuture.allOf(weatherFuture, trafficFuture)
    .thenAccept(v -> renderCityDashboard(weatherFuture.join(), trafficFuture.join()));

🔴 Senior Level

Внутрішня будова: `BiCompletion` та двоспрямована координація

Під капотом операції попарного комбінування використовують вузли, що успадковують клас BiCompletion:

abstract static class BiCompletion<T, U, V> extends UniCompletion<T, V> {
    CompletableFuture<U> snd; // Посилання на другий батьківський ф'ючер
    ...
}

Коли викликається cf1.thenCombine(cf2, fn):
- Створюється вузол BiApply, який реєструється в стеках залежностей одночасно обох ф’ючерів (cf1 та cf2).
- Метод tryFire(mode) перевіряє: чи завершилися обидва джерела (src.result != null && snd.result != null).
- Той потік, який завершить другий (останній) із ф’ючерів, виконає CAS-операцію над вузлом і запустить виконання функції fn.
Усунення витоків пам’яті (Clean-up):
- Щойно один із шляхів активує tryFire, посилання на батьківські ф’ючери зануляються, щоб звільнити пов’язані графи об’єктів у пам’яті.

Обмеження `anyOf` у розподілених архітектурах

Патерн «перегонів» через anyOf приховує в собі небезпеку:

Відсутність автоматичного скасування невдах (No auto-cancellation): після того як найшвидший ф’ючер виграв перегони, решта завдань продовжують виконуватися і витрачати ресурси (потоки, сокети, з’єднання з БД).
Для запобігання витоку ресурсів необхідно явно скасовувати програвших:

public static <T> CompletableFuture<T> firstOfWithCancel(List<CompletableFuture<T>> futures) {
    CompletableFuture<T> promise = new CompletableFuture<>();

    for (CompletableFuture<T> f : futures) {
        f.whenComplete((val, ex) -> {
            if (ex != null) {
                // Якщо потрібна політика "перший успішний", помилку можна проігнорувати,
                // доки не завершаться всі
            } else if (promise.complete(val)) {
                // Скасовуємо всіх інших учасників перегонів!
                futures.forEach(other -> {
                    if (other != f) other.cancel(true);
                });
            }
        });
    }
    return promise;
}

4 Tricky Questions

1. Чому `CompletableFuture.allOf(...)` повертає `CompletableFuture<Void>`, а не типізований список або кортеж результатів?

Відповідь: Метод allOf приймає аргумент змінної довжини з довільними типами: CompletableFuture<?>... cfs. Передані ф’ючери можуть мати абсолютно різні типи даних (наприклад, CompletableFuture<User>, CompletableFuture<Account> та CompletableFuture<Double>). У системі типів Java неможливо статично виразити гетерогенний кортеж невідомого розміру (User, Account, Double). Єдиним спільним типобезпечним знаменником є Void. Тому обов’язок щодо вилучення результатів покладено на розробника через метод .join().

2. Якщо в `CompletableFuture.allOf(cf1, cf2, cf3)` впадуть одразу два ф’ючери з різними винятками, який виняток викине `allOf.join()`? Куди подінеться другий?

Відповідь: Метод allOf.join() викине CompletionException, причиною (cause) якого стане виняток того ф’ючера, який завершився з помилкою першим за часом. Другий виняток у результат allOf не потрапить взагалі і не буде доданий до suppressed! Він залишиться зафіксованим усередині поля result свого власного екземпляра CompletableFuture. Щоб виявити всі помилки в пачці, необхідно явно пройтися по всіх вихідних ф’ючерах і перевірити f.isCompletedExceptionally().

3. Чи скасовує метод `CompletableFuture.anyOf(cf1, cf2)` решту ф’ючерів після того, як перший завершився?

Відповідь: Категорично ні. Специфікація anyOf лише гарантує, що результуючий ф’ючер перейде в завершений стан із результатом першого фінішувавшого завдання. Усі інші $N - 1$ завдань продовжать крутитися в пулі потоків або чекати відповідей по мережі до свого природного завершення. Якщо потрібно перервати їхнє виконання, скасування слід реалізовувати вручну.

4. У якому потоці виконається функція-комбінатор `(a, b) -> a + b` при виклику `cf1.thenCombine(cf2, fn)`?

Відповідь: Функція виконається:

У потоці того ф’ючера, який завершився ОСТАННІМ (оскільки для старту функції потрібна готовність обох).
Або в поточному (викликаючому) потоці, якщо на момент реєстрації thenCombine обидва ф’ючери cf1 та cf2 вже перебували в завершеному стані. Якщо комбінатор виконує блокуючу або важку роботу, необхідно використовувати thenCombineAsync(cf2, fn, dedicatedExecutor).

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

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

Для двох незалежних паралельних завдань: thenCombine (трансформація), thenAcceptBoth (споживання), runAfterBoth (тригер). Затримка = $\max(t_1, t_2)$.
Для $N$ завдань: CompletableFuture.allOf() (чекає на всіх, повертає Void), CompletableFuture.anyOf() (чекає на першого, повертає Object).
Вилучення даних з allOf: через futures.stream().map(CompletableFuture::join).toList(). Оскільки allOf гарантує завершення, join() не блокує потік.
anyOf не скасовує завдання, що залишилися.
allOf при падінні кількох завдань показує лише першу за часом помилку.

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

Плутати thenCombine (паралельно, незалежні) та thenCompose (послідовно, залежні).
Вважати, що allOf повертає готовий список значень.
Забувати ізолювати помилки через exceptionally перед передачею ф’ючерів в allOf.

🟢 Junior Level

1. Попарне об’єднання двох незалежних завдань (thenCombine)

2. Очікування завершення множини ($N$) завдань (allOf)

3. Перегони до першої відповіді (anyOf)

🟡 Middle Level

Систематизація методів комбінації

Ідіоматичний патерн збору результатів з allOf у List<T>

Ізоляція помилок при комбінуванні через allOf

🔴 Senior Level

Внутрішня будова: BiCompletion та двоспрямована координація

Обмеження anyOf у розподілених архітектурах