⏱️ Раздел 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 / Reactor).

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