⏱️ Раздел 19 · Вопрос #24

В каких случаях лучше использовать CompletableFuture, а в каких — реактивное программирование

Критический водораздел между CF и Reactive Streams:

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

🟢 Junior Level

Суть различия в двух предложениях

  • CompletableFuture предназначен для обработки одиночного асинхронного результата (скалярного значения: 0 или 1 элемент) в рамках модели «запрос — ответ».
  • Реактивное программирование (Project Reactor, RxJava, Akka/Pekko) спроектировано для непрерывной обработки асинхронных потоков данных ($0 \dots N$ или бесконечное число элементов) с поддержкой управления скоростью передачи (backpressure).
// CompletableFuture: одиночный запрос — одиночный ответ (1 результат)
CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userService.findById(userId));
userFuture.thenAccept(user -> log.info("User loaded: {}", user.getName()));

// Project Reactor (Flux): асинхронный поток событий (0..N результатов)
Flux<StockPrice> priceStream = stockExchangeClient.streamLivePrices("AAPL");
priceStream
    .filter(p -> p.value() > 150.0)
    .subscribe(price -> log.info("Price alert: {}", price));

Простое эмпирическое правило

  1. Нужно асинхронно сделать 2–3 сетевых вызова к внешним API, параллельно сходить в БД и скомбинировать результат в один DTO $\rightarrow$ CompletableFuture (встроен в JDK, не требует сторонних зависимостей, прост в чтении).
  2. Требуется потоковая обработка (Server-Sent Events, WebSockets, непрерывные метрики, выгрузка миллионов строк из реактивного драйвера БД) или продюсер генерирует данные быстрее, чем консьюмер успевает их обработать $\rightarrow$ Реактивное программирование.

🟡 Middle Level

Архитектурные отличия: Eager vs Lazy, Пуш vs Обратное давление

1. Мощность множества (Cardinality)

  • CompletableFuture<T>: строго одно будущее значение либо исключение. После завершения его состояние неизменно (immutable). Невозможно выразить «поток данных во времени».
  • Mono<T>: $0$ или $1$ элемент (ближайший реактивный аналог CompletableFuture).
  • Flux<T>: $0 \dots \infty$ элементов.

2. Модель исполнения: Eager (Hot) vs Lazy (Cold)

  • CompletableFuture всегда «горячий» (Hot / Eager). Вычисление начинает выполняться сразу в момент вызова CompletableFuture.supplyAsync(...) или передачи задачи в Executor, независимо от того, подписался ли кто-то на результат через thenApply или join.
  • Реактивные пайплайны по умолчанию «холодные» (Cold / Lazy). Декларативное описание пайплайна (flux.map(...).filter(...)) ничего не выполняет, пока на него явно не вызовут .subscribe() (или фреймворк типа Spring WebFlux не сделает это за вас). Поток можно перезапускать заново для каждого нового подписчика.

3. Backpressure (Обратное давление)

Критический водораздел между CF и Reactive Streams:

[Producer] ---- (100 000 элементов/сек) ---> [Queue] ---> [Consumer] (1 000 элементов/сек)
                                                |
                 В CompletableFuture:            | В Reactive Streams (pull-push):
                 Очередь растёт -> OOM!         | Consumer шлет request(10)
                                                | Producer замедляется
  • В CompletableFuture консьюмер не может сказать источнику: «Притормози, отдай мне только следующие 10 элементов». Источник безудержно пушит задачи в ThreadPoolExecutor. Если консьюмер медленнее продюсера, очередь задач переполняется, вызывая RejectedExecutionException либо OutOfMemoryError.
  • В Reactive Streams действует протокол Subscription.request(n): консьюмер сам запрашивает у источника порцию элементов, которую готов обработать.

4. Распространение отмены (Cancellation Propagation)

  • В CompletableFuture.cancel(mayInterruptIfRunning) отмена локальна: завершает текущий узел графа с CancellationException. Вычислительный поток при этом не прерывается, а вышележащие этапы (upstream) не оповещаются об отмене.
  • В Reactive Streams вызов Subscription.cancel() каскадно поднимается вверх по всей цепочке операторов (upstream), останавливает генератор, освобождает TCP-сокеты, отменяет таймауты и освобождает ресурсы.

Сравнительная таблица

Критерий CompletableFuture Project Reactor / RxJava
Единица данных Одно скалярное значение ($T$) Поток событий ($0 \dots \infty$)
Стандарт Java Core (JDK 8+) Reactive Streams Specification (Flow в Java 9)
Модель выполнения Hot (Eager) Cold (Lazy) по умолчанию
Backpressure Отсутствует Встроен на уровне протокола (request(n))
Отмена (Cancellation) Локальная, не прерывает upstream Каскадная вверх по графу (upstream)
Богатство операторов Около 50 методов Более 200 операторов (временные окна, буферы, retry)
Сложность отладки Средняя Высокая (потеря стектрейсов, сложности контекста)

🔴 Senior Level

Эволюция архитектуры: Reactive vs CompletableFuture vs Virtual Threads (Java 21+)

Исторически реактивное программирование внедрялось в Java по двум главным причинам:

  1. Неблокирующий I/O (Non-blocking / Event-loop) для утилизации системных ресурсов: сохранение 1 потока ОС на тысячу сетевых соединений вместо 1 потока на соединение (Thread-per-request).
  2. Потоковая композиция и Backpressure для обработки стримов событий.

С релизом Java 21 (Project Loom / Virtual Threads) баланс кардинально изменился:

// Традиционный Spring WebFlux (Reactor):
public Mono<DashboardDto> getDashboardReactive(Long userId) {
    return userClient.getUser(userId)
        .zipWith(orderClient.getOrders(userId))
        .map(tuple -> new DashboardDto(tuple.getT1(), tuple.getT2()));
}

// Современная Java 21 (Virtual Threads + Structured Concurrency):
public DashboardDto getDashboardVirtualThreads(Long userId) throws InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Supplier<User> userSub = scope.fork(() -> userClient.getUserBlocking(userId));
        Supplier<List<Order>> ordersSub = scope.fork(() -> orderClient.getOrdersBlocking(userId));

        scope.join().throwIfFailed(); // Блокировка виртуального потока дешева (~1 КБ стека)

        return new DashboardDto(userSub.get(), ordersSub.get());
    }
}

Почему Virtual Threads вытесняют Reactive в 80% типовых Enterprise-задач

  1. Сохранение императивного стиля: Снова работают стандартные try-catch, обычные for-циклы, отладчики IDE, привычный ThreadLocal и читаемые стектрейсы без реактивных обёрток.
  2. Устранение “Coloring Problem” (красно-синие функции): В реактивном стеке возврат Mono/Flux заражает все слои приложения — от контроллера до репозитория. В модели Virtual Threads любой метод остаётся обычным синхронным вызовом.

Ниша, где Reactive остаётся незаменимым

Виртуальные потоки решают проблему дешевизны ожидания I/O, но они не умеют управлять скоростью потока данных. Реактивный стек остаётся предпочтительным при:

  1. Необходимости Backpressure в высоконагруженных шлюзах (API Gateway, Rate Limiting, обработка очередей с контролем памяти консьюмера).
  2. Сложной временной математике над потоками: скользящие временные окна (window(Duration)), объединение быстрых и медленных потоков событий во времени (sample, debounce, throttle), стриминг котировок, IoT-телеметрия.
  3. Долгоживущих дуплексных соединениях с потоковой отдачей: Server-Sent Events (SSE), WebSockets, gRPC Streaming.

4 Tricky Questions

1. Что произойдет, если источник выдаст 1 000 000 элементов в секунду в CompletableFuture vs Reactive Stream?

Ответ: В архитектуре на CompletableFuture каждый элемент потребует планирования задачи в пул потоков (например, через thenApplyAsync). Так как консьюмер не имеет механизма торможения продюсера (нет backpressure), внутренняя очередь пула потоков начнет стремительно расти. Это неизбежно приведет либо к отказу в обслуживании (RejectedExecutionException), либо к катастрофическому исчерпанию кучи (OutOfMemoryError: Java heap space).

В реактивном потоке (Flux / Flow.Publisher) консьюмер изначально запрашивает только то количество элементов, которое успевает обработать (например, request(32)). Продюсер приостанавливает генерацию/чтение из сокета на уровне TCP receive buffer, пока консьюмер не потребит порцию и не запросит следующую. Если источник неуправляем (горячий поток датчиков), реактивный пайплайн позволяет явно задать стратегию переполнения: onBackpressureDrop() (отбрасывать излишки), onBackpressureLatest() (хранить только последнее значение) или буфер с ограничением размера.

2. Является ли CompletableFuture «горячим» (hot) или «холодным» (cold) источником по сравнению с Reactor Mono?

Ответ: CompletableFuture является строго «горячим» (Hot). Он представляет собой обещание результата вычисления, которое уже запущено (или завершено). Если вы создали экземпляр CompletableFuture.supplyAsync(task), задача уже отправлена в executor, независимо от того, прикрепите ли вы к ней коллбэки. Если три разных метода вызовут .thenApply() на одном и том же CompletableFuture, фоновая задача выполнится ровно один раз, а результат будет доставлен всем троим подписчикам.

Mono в Project Reactor по умолчанию «холодный» (Cold). Конструкция Mono.fromCallable(task) или Mono.defer(...) лишь объявляет рецепт вычисления. Если на него никто не вызовет .subscribe(), фоновая работа вообще никогда не начнется. Если на такой холодный Mono подпишутся три независимых консьюмера, задача выполнится три раза (по разу для каждого подписчика), если поток явно не сделан горячим через .share() или .publish().autoConnect().

3. Как работает отмена (cancellation) в CompletableFuture vs Reactive Streams? Можно ли остановить уже запущенный HTTP-запрос?

Ответ: В CompletableFuture.cancel(true) семантика отмены крайне ограничена:

  • Флаг mayInterruptIfRunning игнорируется в стандартной реализации JDK.
  • Вызов cancel() просто записывает в текущую стадию специальный внутренний результат AltResult(CancellationException). Поток, выполняющий сетевой запрос или вычисление, не прерывается и продолжает работать до физического завершения.
  • Никакой сигнал отмены не идет «вверх» к источнику данных.

В Reactive Streams отмена является фундаментальной частью протокола. Когда подписчик вызывает Subscription.cancel(), сигнал проходит каскадом вверх (upstream) по всей цепочке операторов. Каждый оператор закрывает свои буферы, таймеры и ресурсы, а сам источник (например, реактивный WebClient на базе Netty) закрывает сетевой сокет/канал TCP, мгновенно прекращая чтение данных и освобождая сетевой буфер ОС.

4. Если в Java 21 появились Virtual Threads, полностью ли они вытесняют реактивное программирование и CompletableFuture?

Ответ: Нет, не полностью. Virtual Threads закрывают главную боль классической Java — они делают блокирующий I/O дешевым по памяти и переключению контекста. Поэтому в классических веб-приложениях типа «получил HTTP-запрос $\rightarrow$ сходил в базу $\rightarrow$ дернул REST API $\rightarrow$ отдал JSON» виртуальные потоки (в связке со стандартным синхронным кодом или StructuredTaskScope) делают реактивный стек избыточным усложнением.

Однако реактивное программирование сохраняет свою незаменимость в следующих областях:

  1. Стриминг данных с контролем скорости (Backpressure): виртуальные потоки не предоставляют протокола регуляции скорости между независимыми потоками продюсера и консьюмера.
  2. Сложная временная обработка событий: реактивные операторы типа debounce (подавление дребезга), sample, bufferTimeout, window сложно и неэффективно писать на чистых потоках с блокировками.
  3. Реактивные протоколы: RSocket, gRPC Streaming, WebSocket, SSE.

🎯 Шпаргалка для интервью

Матрица выбора архитектурного решения

                             [ Какова природа задачи? ]
                                         |
            +----------------------------+----------------------------+
            |                                                         |
 [ Одиночный запрос-ответ ]                                [ Поток данных (0..N) ]
            |                                                         |
   Java 21+ доступна?                                      Нужен ли Backpressure?
    /              \                                        /                  \
 [Да]              [Нет]                                  [Да]                 [Нет]
  |                  |                                      |                    |
Синхронный код     CompletableFuture                Project Reactor         Обычный Stream /
на Virtual Threads  (или CF + Custom Executor)       (Flux / Mono)          Channel / Queue

Ключевые тезисы для собеседования

  1. Мощность: CF — это скаляр ($0..1$). Реактивный стрим (Flux) — это вектор/поток ($0..\infty$).
  2. Backpressure: В CF отсутствует полностью (опасность OOM). В Reactive Streams это фундаментальная часть спецификации (Subscription.request(n)).
  3. Eager vs Lazy: CF горячий (стартует сразу, повторный опрос берет кэшированный результат). Реактивный пайплайн холодный (не запустится без .subscribe(), перезапускается на каждую подписку).
  4. Virtual Threads: Не убивают реактивность, но сужают ее нишу до реального стриминга, шин событий и систем с жесткими требованиями к backpressure.

Антипаттерны (Красные флаги на собеседовании)

  • ❌ «Мы переписали CRUD-сервис на Spring WebFlux, чтобы он работал в 10 раз быстрее» — WebFlux не ускоряет одиночные вызовы; он лишь повышает плотность соединений при медленном I/O ценой усложнения стектрейсов и потери отлаживаемости.
  • ❌ «CompletableFuture поддерживает обратное давление, если указать ограниченную очередь в ThreadPoolExecutor» — это не backpressure, а грубый отказ в обслуживании (RejectedExecutionHandler).
  • ❌ «Virtual Threads делают реактивное программирование устаревшим мусором» — Virtual Threads решают проблему блокировок, но не дают операторов временных окон, объединения потоков и backpressure.

Связанные темы