Как комбинировать результаты нескольких 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 в распределённых архитектурах