CompletableFuture в Java для составных асинхронных операций против тупиковых цепочек без ясного контекста исполнения

Цепочка из supplyAsync, thenApply и thenAccept читается как конвейер и вызывает заслуженную гордость у автора. На экране ступени соединяются стрелками, каждая строчка выглядит асинхронной, и код занимает вдвое меньше места, чем ручная работа с потоками. Настоящая асинхронность живёт не в строчках, а в ответе на один вопрос: каким потоком исполняется каждый шаг. Пока ответ известен, ступени собираются в составные операции, которые честно экономят время. Когда ответ теряется, та же красота превращает сервис в шахматную партию с самим собой, где каждый ход может закончиться самозапиранием. Ниже разбор того, как CompletableFuture распределяет работу, как строить веера из независимых вызовов и где проходят границы разумного применения.

Базовые ступени CompletableFuture от источника к терминальной операции

Класс появился в восьмом выпуске Java в марте 2014 года и принёс в стандартную библиотеку функциональную модель композиции стадий. Источник создаётся парой методов: supplyAsync принимает поставщика и возвращает будущий результат, runAsync исполняет действие без результата. Дальше идут переходы: thenApply превращает значение, thenAccept потребляет его, thenCompose склеивает две последовательные зависимости, избегая футуры в футуре. Каждая ступень возвращает новый объект, и конвейер читается слева направо:

CompletableFuture<Integer> estimate = CompletableFuture
        .supplyAsync(() -> fetchPrice("sku-42"))
        .thenApply(p -> p * (100 - discount) / 100)
        .thenCompose(p -> fetchDeliveryEstimate(p));

Ступени иммутабельны, каждый промежуточный объект можно передать, скомбинировать и переиспользовать, не боясь, что чужой код перепишет уже собранное. Имена методов образуют маленький словарь, и после недели работы с ним чтение чужой цепочки перестаёт требовать словаря настоящего. Рядом с основными переходами в словаре живут обработчики края: exceptionally чинит сбой, handle видит обе стороны, whenComplete ставит наблюдателя. Освоив двадцать имён, разработчик перестаёт замечать, что перед ним не просто вызовы, а граф исполнения с точками присоединения, и именно этот граф, а не строчки, определяет поведение под нагрузкой.

Словарь обманчив в одном месте: он молчит о том, кто именно исполняет тела, и этот вопрос не виден в коде цепочки вовсе.

Кто исполняет ступени, когда исполнитель не указан явно

Суффикс Async в имени метода это обещание, а не свойство. Без суффикса ступень исполняет поток, завершивший предыдущую стадию, а если стадия уже завершена, поток, выполняющий регистрацию. Со суффиксом и без явного исполнителя работа уходит в общий пул ForkJoinPool.commonPool, параллелизм которого равен числу ядер минус один, и в этом пуле же живут параллельные стримы со всей машины.

Первое следствие: thenApply с блокирующим телом асинхронен ровно ноль раз. Если предыдущую стадию завершал поток сетевого клиента, тело исполнится в этом же потоке, и событийный цикл клиента встанет на время обращения к базе. Второе следствие приятней: короткие чистые тела на завершившей стадии не стоят ни переключения контекста, ни очереди, и на этом держится вся скорость немеченых ступеней. Баланс решает разработчик: тяжёлое или блокирующее тело отправляется в именованный исполнитель явно.

CompletableFuture.supplyAsync(this::fetch, ioPool)
                 .thenApplyAsync(this::parseSlow, ioPool)
                 .thenAccept(this::render);

Классическая авария выглядит так. supplyAsync положил задачу в общий пул, внутри неё вызван join на другой футуре, которая ждёт свободного рабочего того же пула. Рабочих нет, все ждут, и сервис молчит без единого сообщения об ошибке. Дамп потоков показывает узнаваемую картину: рабочие общего пула стоят в ожидании самих себя. Лечение одно: блокирующие тела выносить из общего пула в пул с запасом потоков, а вызовы join не оставлять внутри ступеней общего пула.

Вопрос о потоке исполнения проверяется экспериментом в десять строк: в тело ступени кладётся имя текущего потока, и результат печатается. Первая ступень покажет рабочего указанного пула, немеченая ступень отработает в том же рабочем, а присоединённая к уже завершённой стадии исполнится в потоке вызывающего. Десять минут такого опыта экономят дни теории:

CompletableFuture
        .supplyAsync(() -> Thread.currentThread().getName(), ioPool)
        .thenApply(n -> n + " -> " + Thread.currentThread().getName())
        .thenAccept(System.out::println);

Арифметика общего пула безжалостна. На шестнадцати ядрах он держит пятнадцать рабочих, и пятнадцать одновременных обращений к базе по сотне миллисекунд занимают его на весь срок. Шестнадцатое обращение ждёт, а рядом в том же пуле стоит параллельный стрим из соседнего модуля. Отдельный пул ввода-вывода на пятьдесят потоков снимает конфликт одним движением.

Составные операции для независимых вызовов и сборка результата

Настоящая сила класса раскрывается не в линейных цепочках, а в веерах. Три независимых вызова исполняются параллельно и складываются в общий результат:

CompletableFuture<Price> a = CompletableFuture.supplyAsync(() -> getPrice(id), ioPool);
CompletableFuture<Stock> b = CompletableFuture.supplyAsync(() -> getStock(id), ioPool);
CompletableFuture<Rating> c = CompletableFuture.supplyAsync(() -> getRating(id), ioPool);

CompletableFuture<Page> page = a
        .thenCombine(b, (price, stock) -> new PriceBlock(price, stock))
        .thenCombine(c, (block, rating) -> new Page(block, rating));

Время сборки равно времени самого медленного вызова вместо суммы трёх, и на странице с полудюжиной внешних источников это разница между полусекундой и тремя. Для десятков источников служит метод allOf, он возвращает футуру без результата, и значения собирают уже после завершения:

CompletableFuture<Void> all = CompletableFuture.allOf(futures);
all.thenRun(() -> {
    List<Price> done = futures.stream()
            .map(CompletableFuture::join)
            .toList();
});

Экономия веера считается на салфетке. Три последовательных вызова по сто миллисекунд дают триста, три параллельных около ста двадцати с учётом сшивки. С ростом числа ветвей без allOf и явного исполнителя уже не обойтись, и каждая новая ветвь требует свежего взгляда на крайний срок сборки.

Для гонки источников служит anyOf: завершение первой футуры из набора продолжает цепочку, остальные теряют значение. Типовое применение это дубль-запрос к двум репликам, где побеждает быстрейшая, а отставший ответ отправляется в отмену. Композиция не усложняется: любая из ветвей приводит результат к общему типу заранее.

Обратная сторона веера, о которой молчит любая красота: сбой одного источника провалит всю сборку целиком, и причину придётся искать по ветвям. Гигиена веера из трёх правил ставит эту беду под контроль: у каждой ветви свой обработчик сбоя, у всей сборки свой крайний срок, и результат, потерявший один источник, собирается по остатку, если продукт позволяет.

Ошибки и таймауты в длинных цепочках вызовов

Исключение в ступени не выбрасывается вызывающему коду, оно путешествует по цепочке до обработчиков. exceptionally подставляет запасное значение, handle видит и значение, и сбой одновременно, whenComplete наблюдает, не вмешиваясь в поток результата. Обработчики ставят на конкретные места цепочки, а не в конце всего конвейера, иначе ветка с падением останется незамеченной до самого терминала. Место имеет значение: exceptionally, поставленный в середине, чинит сбой для всех последующих ступеней, но не спасает ветви, присоединённые выше места сбоя. Обработчик на конце покрывает всё, зато лишает следующие ступени шанса отработать запасное значение. Выбор места это выбор между спасением ветки и спасением всего графа. Девятый выпуск Java в сентябре 2017 года добавил часовых:

CompletableFuture.supplyAsync(this::fetch, ioPool)
        .orTimeout(300, TimeUnit.MILLISECONDS)
        .completeOnTimeout(DEFAULT_VALUE, 300, TimeUnit.MILLISECONDS);

Первый метод завершает футуру сбоем по истечении срока, второй подставляет значение по умолчанию. До них таймауты приходилось собирать вручную на планировщике, и каждый такой узел был местом утечки. Соседний механизм delayedExecutor строит отложенные ступени на том же планировщике без ручных ScheduledExecutorService.

Правило гигиены короткое. У каждой внешней ветви свой крайний срок, у каждой цепочки свой обработчик сбоев, и ни одна футура не остаётся без надзора: незакрытая ветвь держит занятого рабочего и тащит в продакшен тайну, разрешающуюся по nobody-знает-какому графику.

Виртуальные потоки и место CompletableFuture в новых проектах

Двадцать первый выпуск Java в сентябре 2023 года закрепил виртуальные потоки, и дешёвая блокировка изменила расстановку. Код, который раньше собирался из асинхронных ступеней, теперь пишется в прямом стиле: запустил десятки виртуальных потоков, вызвал join, получил результат. Фан-аут из сотни обращений читается проще любой цепочки, и часть нагрузок ушла с CompletableFuture окончательно.

ExecutorService virtual = Executors.newVirtualThreadPerTaskExecutor();
CompletableFuture.supplyAsync(this::fetch, virtual)
                 .thenApplyAsync(this::parse, virtual);

Класс не уходит со сцены и после этого. Он остаётся естественным языком композиции для библиотечных интерфейсов, где внутренности скрыты от вызывающего, для вееров с общим крайним сроком и для мостиков между подсистемами с разными моделями исполнения. Механизмы структурной конкурентности, которые идут в выпусках JDK в статусе предпросмотра, доводят идею до конца: у параллельных подзадач появляется владелец, а у группы единый крайний срок, и забытые ветви перестают существовать как класс.

Асинхронность по своей природе сильнее прямого стиля там, где событие приходит извне: сетевые клиенты на событийных циклах, очереди сообщений, потоки уведомлений. Там CompletableFuture работает адаптером между чужой событийной моделью и привычным графом ступеней, и виртуальные потоки не заменяют такой адаптер, а лишь снабжают его телом.

Границы применимости и выбор подхода под конкретную задачу

Вопросы из списка ниже работают и как ревью-чеклист. Разработчик, способный ответить на все пять применительно к своей цепочке, уже не наращивает конвейер вслепую, и его граф ступеней читается как проект, а не как загадка.

Выбор между цепочкой, веером и прямым стилем сводится к пяти вопросам:

1. есть ли в теле ступени блокирующая работа, и если есть, ступень уходит в именованный исполнитель, а не в общий пул;
2. есть ли у операции крайний срок, и если нет, ставится orTimeout или явный механизм с запасным значением;
3. независимы ли вызовы, и если независимы, сборка идёт веером через thenCombine или allOf, а не наивной последовательностью;
4. спроектирован ли отказ одной ветви так, чтобы сборка продолжалась по остатку, а не проваливалась целиком;
5. не проще ли задача для прямого блокирующего кода на виртуальных потоках, который читается и меняется легче любой цепочки.

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

Практика команд, живущих на этом классе годами, сходится в одном: цепочки держат дисциплина и диагностика, а не самоконтроль отдельного автора. Именованные исполнители видны в конфигурации, у каждой ветви есть метрика фактического времени, а график ступеней сопровождается схемой потоков, которую рисуют один раз и обновляют при каждом изменении конвейера. Те же операции в беспечных руках собирают обратное: общий пул, забитый чужими блокировками, ветвь без таймаута, ждущая уже исчезнувший сервис, и веер, падающий от единственного сбоя. Разница между двумя картинками не в классе и не в версии JDK, а в привычке задавать вопрос о потоке исполнения для каждой ступени до того, как цепочка поедет в продакшен. Инструмент здесь ни в чём не виноват, виновата инерция: конвейер, который вчера был длиной в три ступени, сегодня дотягивается до чужого пула, и заметить это удобнее на бумаге, чем в дампе потоков ночью.