CompletableFuture: асинхронные задачи в Java

← назад к разделу

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

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

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

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

До Java 8 асинхронную задачу представлял Future, и умел он немного: проверить, готово ли (isDone()), заблокироваться в ожидании (get() — в том числе с потолком, get(5, TimeUnit.SECONDS)) и попробовать отменить (cancel(), isCancelled()). А вот выстроить цепочку из нескольких шагов на нём было нечем. CompletableFuture — расширенный Future: он реализует интерфейс CompletionStage, и на этом держатся цепочки обработки, объединение нескольких задач и обработка ошибок без try-catch вокруг get().

Запуск задачи: supplyAsync и runAsync

Самый простой способ запустить задачу асинхронно — supplyAsync. Он принимает Supplier<T> и сразу возвращает CompletableFuture<T>, не дожидаясь результата:

живой пример

import java.util.concurrent.CompletableFuture;

public class AsyncStartDemo {
    public static void main(String[] args) throws Exception {
        CompletableFuture<String> user = CompletableFuture.supplyAsync(
            () -> "Анна, найдена в потоке " + Thread.currentThread().getName());

        System.out.println("главный поток не ждёт: " + Thread.currentThread().getName());
        System.out.println(user.get());
    }
}

Главный поток напечатал свою строку, не дожидаясь задачи, и заблокировался только на get() — там, где результат действительно понадобился. Если возвращать нечего, есть runAsync: он принимает Runnable и отдаёт CompletableFuture<Void>.

На каком пуле выполняется задача

Во второй строке вывода видно имя чужого потока — обычно ForkJoinPool.commonPool-worker-1. По умолчанию CompletableFuture использует общий пул ForkJoinPool.commonPool(), один на всю JVM. Для коротких вычислений это нормально, но для операций ввода-вывода (запросы к базе, HTTP-вызовы) лучше передавать собственный Executor:

ExecutorService ioPool = Executors.newFixedThreadPool(20);

CompletableFuture<String> future = CompletableFuture.supplyAsync(
    () -> fetchUserFromDatabase(userId),
    ioPool  // задача уйдёт именно в этот пул
);

Почему важно разделять? Если заблокировать все потоки commonPool ожиданием сетевых ответов, другие CompletableFuture в JVM встанут в очередь — включая те, что вообще не связаны с вашим кодом.

Цепочки обработки: thenApply и thenAccept

Получив результат асинхронной задачи, вы часто хотите что-то с ним сделать: получили заказ, посчитали сумму, показали пользователю. Для этого есть операторы-трансформации, и различаются они тем, что остаётся после шага. thenApply(Function<T, R>) преобразует результат и возвращает новый CompletableFuture<R>, это «посчитать сумму». thenAccept(Consumer<T>) потребляет результат и возвращает CompletableFuture<Void>, это «показать». thenRun(Runnable) запускает действие после завершения, не глядя на результат, это «записать в журнал, что шаг прошёл».

живой пример

import java.util.concurrent.CompletableFuture;

public class ChainDemo {
    public static void main(String[] args) {
        String email = CompletableFuture.supplyAsync(() -> "Anna Ivanova")
            .thenApply(name -> name.split(" ")[0])
            .thenApply(first -> first.toLowerCase() + "@example.com")
            .join();

        System.out.println(email);   // anna@example.com
    }
}

Каждый thenApply создаёт новый этап цепочки. Если к моменту вызова предыдущий шаг ещё не закончился, этап выполнит тот поток, который его завершит. А если future к этому моменту уже готов, этап выполнится прямо здесь — в потоке, который вызвал thenApply. Гарантии «всегда в фоне» нет ни в одном из двух случаев, и при отладке это важно помнить: строчка из середины цепочки вполне может напечататься с именем main. Если хотите, чтобы следующий этап гарантированно выполнился в отдельном потоке из пула, используйте вариант с суффиксом Async:

.thenApplyAsync(user -> heavyTransformation(user), ioPool)

Композиция задач: thenCompose

Представьте, что по результату первой асинхронной задачи нужно запустить вторую — тоже асинхронную. С thenApply получится CompletableFuture<CompletableFuture<T>> — future внутри future, который придётся разворачивать дважды. thenCompose разворачивает цепочку сам:

живой пример

import java.util.concurrent.CompletableFuture;

public class ComposeDemo {
    public static void main(String[] args) {
        CompletableFuture<Integer> user = CompletableFuture.supplyAsync(() -> 42);

        var nested = user.thenApply(id -> CompletableFuture.supplyAsync(() -> "заказ " + id));
        var flat = user.thenCompose(id -> CompletableFuture.supplyAsync(() -> "заказ " + id));

        System.out.println(nested.join().getClass().getSimpleName());  // снова CompletableFuture
        System.out.println(flat.join());                               // заказ 42
    }
}

Короткая формула: thenApply — когда следующий шаг синхронный; thenCompose — когда следующий шаг сам возвращает CompletableFuture.

Объединение результатов: thenCombine и allOf

Иногда нужно запустить две задачи параллельно и объединить их результаты. thenCombine ждёт завершения обоих CompletableFuture и применяет функцию:

живой пример

import java.util.concurrent.CompletableFuture;

public class CombineDemo {
    public static void main(String[] args) {
        CompletableFuture<String> user = CompletableFuture.supplyAsync(() -> "Анна");
        CompletableFuture<Integer> wallet = CompletableFuture.supplyAsync(() -> 1500);

        String line = user.thenCombine(wallet, (u, w) -> u + " — баланс: " + w).join();

        System.out.println(line);                                            // Анна — баланс: 1500
        System.out.println(CompletableFuture.allOf(user, wallet).isDone());  // true
    }
}

Обе задачи выполняются параллельно, результат собирается только когда обе готовы. anyOf возвращает результат первой завершившейся — это и есть «кто первый ответит».

А allOf сам по себе бесполезен: он возвращает CompletableFuture<Void>, то есть говорит «все готовы», но не отдаёт результаты. Пользуются им всегда одинаково — дожидаются, а потом забирают значения из исходных future, где они уже лежат:

List<CompletableFuture<Price>> futures = ids.stream()
    .map(id -> CompletableFuture.supplyAsync(() -> fetchPrice(id), ioPool))
    .toList();

CompletableFuture<List<Price>> all = CompletableFuture
    .allOf(futures.toArray(CompletableFuture[]::new))
    .thenApply(v -> futures.stream().map(CompletableFuture::join).toList());

join() внутри thenApply не блокирует: к этому моменту allOf уже дождался всех, и каждое значение на месте. Это единственный рабочий способ собрать список результатов из списка задач, и его стоит запомнить целиком. Оговорка: если упала хотя бы одна задача, allOf завершится с ошибкой, а join() на первой же упавшей бросит CompletionException; чтобы собрать частичный результат, каждой задаче дают свой exceptionally до попадания в список.

Обработка ошибок: exceptionally и handle

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

exceptionally(Function<Throwable, T>) — обработчик вызывается только при ошибке, в норме шаг прозрачно пропускается:

живой пример

import java.util.concurrent.CompletableFuture;

public class FailingDemo {
    public static void main(String[] args) {
        String email = CompletableFuture.<String>supplyAsync(() -> {
                throw new IllegalStateException("база недоступна");
            })
            .thenApply(value -> {
                System.out.println("этот шаг не выполнится");
                return value.toLowerCase();
            })
            .exceptionally(ex -> "запасной адрес, причина: " + ex.getCause().getMessage())
            .join();

        System.out.println(email);
    }
}

Строка «этот шаг не выполнится» так и не напечаталась. И ещё деталь: когда исключение вылетело внутри стадии, в обработчик приходит не оно само, а обёртка CompletionException — настоящая причина лежит в getCause().

Но так бывает не всегда. Если future завершили руками, через completeExceptionally(ex), в exceptionally придёт ровно тот ex, который передали, безо всякой обёртки, и getCause() вернёт null. Строчка вида ex.getCause().getMessage() на таком future упадёт с NullPointerException — прямо внутри обработчика ошибок. Надёжный способ достать причину один:

Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;

handle(BiFunction<T, Throwable, R>) — вызывается всегда, независимо от успеха или ошибки; один из аргументов будет null:

.handle((user, ex) -> {
    if (ex != null) {
        return fallbackUser();
    }
    return user;
})

Таймауты, отмена и whenComplete

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

Таймаут. Сам по себе CompletableFuture ждёт вечно: если сосед не ответил, цепочка не завершится никогда. С Java 9 есть два метода — orTimeout превращает просрочку в ошибку, completeOnTimeout подставляет запасное значение:

fetchPriceAsync(id)
    .orTimeout(2, TimeUnit.SECONDS)                    // не ответил — TimeoutException
    .exceptionally(ex -> Price.unknown());

fetchRecommendationsAsync(userId)
    .completeOnTimeout(List.of(), 300, TimeUnit.MILLISECONDS);   // не успел — отдаём пустой список

Важная оговорка: таймаут завершает future, а не работу. Поток, который висит на медленном запросе, продолжит висеть, соединение останется занятым — просто его результат уже никому не нужен. Поэтому таймаут на клиенте (HTTP, JDBC) всё равно нужен, а orTimeout это страховка сверху и граница для вызывающего кода.

Отмена. cancel(true) у CompletableFuture ведёт себя не так, как у Future из пула, и это регулярно застаёт врасплох: он не прерывает выполняющуюся задачу, а лишь переводит future в отменённое состояние, чтобы следующие стадии не выполнялись и join() бросил CancellationException. Аргумент mayInterruptIfRunning игнорируется. Значит, отменить уже начатый запрос к базе через CompletableFuture нельзя: нужен либо таймаут на самом клиенте, либо своя проверка флага внутри задачи.

whenComplete против handle. Оба вызываются и при успехе, и при ошибке, а разница в том, что остаётся после них. handle подменяет результат тем, что вернула ваша функция: ошибку можно превратить в значение и погасить. whenComplete результат не меняет — он только смотрит: пришло значение, пришла ошибка, и дальше по цепочке едет ровно то же самое, включая ошибку.

future.whenComplete((value, ex) -> log.info("шаг закончился, ошибка={}", ex))   // ошибка поедет дальше
      .handle((value, ex) -> ex != null ? fallback() : value);                  // а здесь погашена

Отсюда правило: whenComplete для побочных действий (лог, метрика, освобождение ресурса), handle и exceptionally — для решения, что вернуть.

Грабли: блокирующий get и проглоченные исключения

Ловушка 1: блокирующий get

Вызов .get() блокирует текущий поток до готовности результата. Внутри стадии, которая сама выполняется в потоке пула, это значит: поток занят ожиданием вместо работы, а задачи, ждущие своей очереди, стоят.

// так не надо: стадия блокирует поток пула ради вложенного future
CompletableFuture.supplyAsync(() -> fetchUser(userId))
    .thenApply(user -> fetchEmailAsync(user).join());

// так надо: вложенный future подставляет thenCompose, ждать не приходится
CompletableFuture.supplyAsync(() -> fetchUser(userId))
    .thenCompose(user -> fetchEmailAsync(user));

В плохом варианте стоит .join(), а не .get(), и не случайно: get() объявляет проверяемые InterruptedException и ExecutionException, а Function их не пропускает — без try-catch код не скомпилируется. .join() бросает непроверяемое CompletionException и потому в цепочках встречается чаще. Только выбор между ними здесь ни при чём: плохо само ожидание внутри стадии. Заменить .join() на .get() с try-catch — значит оставить ту же ошибку, просто записать её длиннее. Ждать надо не внутри цепочки, а на верхнем уровне, за пределами пула; внутри цепочки вложенный future подставляют через thenCompose.

Ловушка 2: проглоченные исключения

Если к CompletableFuture не подключить обработчик ошибок и нигде не вызвать .get() или .join() — исключение просто пропадает. Особенно часто это случается с runAsync:

живой пример

import java.util.concurrent.CompletableFuture;

public class SwallowedDemo {
    public static void main(String[] args) throws Exception {
        CompletableFuture<Void> task = CompletableFuture.runAsync(() -> {
            throw new RuntimeException("что-то сломалось");
        });

        Thread.sleep(200);
        System.out.println("в консоли ни строчки про ошибку");
        System.out.println("а задача провалилась: " + task.isCompletedExceptionally());
    }
}

Правило: всегда завершайте цепочку либо .exceptionally(), либо .handle(), либо явным .join() там, где ошибку можно поймать.

Ловушка 3: общий пул мал, особенно в контейнере

Размер ForkJoinPool.commonPool() — число доступных процессоров минус один, но не меньше одного. JVM видит лимиты контейнера: у пода с лимитом в 3 CPU в общем пуле останется два рабочих потока. Любая блокирующая операция в supplyAsync без executor'а — запрос к базе, HTTP — выстроит задачи в эту пару по очереди, и обработчики пойдут по два, хотя база отвечает быстро.

Порог здесь простой: общий пул берётся, только если ForkJoinPool.getCommonPoolParallelism() > 1. На лимите в 1-2 CPU параллелизм падает до единицы, и тогда CompletableFuture общий пул не берёт вовсе — он подставляет внутренний executor, а тот заводит новый поток на каждую задачу. Очереди нет, но нет и предела: потоков столько, сколько задач, а в выводе вместо ForkJoinPool.commonPool-worker-1 стоит Thread-0.

Правило: блокирующие операции — только в свой executor (или виртуальные потоки), переданный в supplyAsync и *Async-варианты стадий; общий пул — для коротких вычислений.

Контекст между стадиями теряется

Последнее, обо что спотыкаются уже в проде. Стадии цепочки выполняются в разных потоках, а traceId в логах, контекст безопасности и текущая транзакция живут в ThreadLocal того потока, который принял запрос. Значит, всё это в стадиях пропадает: логи асинхронного шага не привязаны к запросу, SecurityContextHolder пуст, @Transactional вызывающего не продолжается.

Лечение то же, что у @Async: переносить контекст явно, в точке передачи.

Map<String, String> mdc = MDC.getCopyOfContextMap();
CompletableFuture.supplyAsync(() -> {
    MDC.setContextMap(mdc);
    try {
        return fetchPrice(id);
    } finally {
        MDC.clear();                 // поток вернётся в пул, чужой traceId в нём не нужен
    }
}, ioPool);

Делать это в каждой стадии руками невозможно, поэтому контекст переносят один раз — декоратором задач у своего пула (ThreadPoolTaskExecutor.setTaskDecorator в Spring) или обёрткой Executor, которая копирует нужные ThreadLocal вокруг каждой задачи. Что именно не переезжает и как это выглядит в Spring, разбирает статья про многопоточность в Spring.

Коротко

• CompletableFuture — асинхронная задача с поддержкой цепочек и обработки ошибок, в отличие от старого Future.
• supplyAsync / runAsync — запуск задачи; без Executor-аргумента — в commonPool, для ввода-вывода лучше передавать свой пул.
• thenApply — трансформация результата (синхронный следующий шаг); thenCompose — когда следующий шаг сам возвращает CompletableFuture.
• thenCombine — объединить два параллельных результата, allOf — дождаться всех, anyOf — взять первый готовый.
• exceptionally — обработчик ошибки с подстановкой значения, handle — обработчик, вызываемый всегда; ошибка из стадии приходит обёрнутой в CompletionException, а переданная в completeExceptionally — как есть, поэтому причину достают через проверку instanceof.
• Две главные грабли: блокирующий .get() внутри потока пула — поток простаивает вместо работы; цепочка без обработчика — исключение теряется молча.
• allOf возвращает Void: список результатов собирают как allOf(...).thenApply(v -> futures.stream().map(CompletableFuture::join).toList()).
• Без orTimeout или completeOnTimeout цепочка ждёт вечно; таймаут завершает future, но не саму работу, поэтому таймаут на клиенте всё равно нужен.
• cancel(true) у CompletableFuture не прерывает выполняющуюся задачу, а только отменяет future; whenComplete смотрит и пропускает ошибку дальше, handle подменяет результат.
• Между стадиями теряются MDC, контекст безопасности и транзакция: их переносят декоратором задач у своего пула.

Что почитать дальше

• ExecutorService и пулы потоков — как устроены пулы, на которых работают асинхронные задачи.
• Виртуальные потоки — альтернатива CompletableFuture для высококонкурентного ввода-вывода в Java 21.
• Структурированная конкурентность — как не потерять параллельные подзадачи, когда их становится много.
• Типичные ошибки многопоточности — дедлоки, гонки и другие проблемы, которые легко допустить в асинхронном коде.