CompletableFuture

Future из прошлого урока умеет только одно — ждать результат через get(). CompletableFuture идёт дальше: к будущему результату можно заранее прицепить следующие шаги — «когда загрузится, преобразуй», «когда оба готовы, объедини». Такой код называют асинхронным: задачи выполняются в фоне, а главный поток не простаивает в ожидании каждой из них.

Первая асинхронная задача

CompletableFuture.supplyAsync запускает задачу в фоновом потоке и сразу возвращает объект-обещание. Метод join() ждёт результат:

void main() {
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 6 * 7);
 
    IO.println("Результат: " + future.join());
}

Результат:

Результат: 42

join() похож на get() из Future, но не требует обрабатывать проверяемые исключения. По умолчанию задачи выполняются в общем пуле потоков ForkJoinPool.commonPool().

Цепочка шагов

thenApply преобразует результат, когда он будет готов, — как map у потоков данных. thenAccept получает итоговое значение и ничего не возвращает:

void pause(long millis) {
    try {
        Thread.sleep(millis);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}
 
String loadName(int userId) {
    pause(200);
    return "анна";
}
 
void main() {
    CompletableFuture<Void> done = CompletableFuture
            .supplyAsync(() -> loadName(7))
            .thenApply(name -> name.substring(0, 1).toUpperCase() + name.substring(1))
            .thenApply(name -> "Здравствуйте, " + name + "!")
            .thenAccept(text -> IO.println(text));
 
    IO.println("Ждём загрузку...");
    done.join();
}

Результат:

Ждём загрузку...
Здравствуйте, Анна!

Строка «Ждём загрузку...» вывелась первой: главный поток не ждал, пока фоновая задача загрузит имя. Метод pause нужен, чтобы спрятать проверяемое InterruptedException от Thread.sleep.

Объединение двух задач

thenCombine дожидается двух независимых задач и объединяет их результаты. Задачи при этом выполняются параллельно:

void pause(long millis) {
    try {
        Thread.sleep(millis);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}
 
void main() {
    CompletableFuture<Double> price = CompletableFuture.supplyAsync(() -> {
        pause(300);
        return 20.0;
    });
    CompletableFuture<Double> rate = CompletableFuture.supplyAsync(() -> {
        pause(300);
        return 0.9;
    });
 
    double total = price.thenCombine(rate, (p, r) -> p * r).join();
    IO.println("Цена в евро: " + total);
}

Результат:

Цена в евро: 18.0

Обе задачи по 300 мс шли одновременно, поэтому итог готов примерно через 300 мс, а не 600.

Много задач

CompletableFuture.allOf возвращает обещание, которое выполнится, когда завершатся все переданные задачи. После этого результаты можно собрать через join() в нужном порядке:

void main() {
    List<String> cities = List.of("Рим", "Осло", "Лима");
 
    List<CompletableFuture<String>> futures = cities.stream()
            .map(city -> CompletableFuture.supplyAsync(() -> city + ": " + city.length() * 5 + "°"))
            .toList();
 
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
 
    for (CompletableFuture<String> f : futures) {
        IO.println(f.join());
    }
}

Результат:

Рим: 15°
Осло: 20°
Лима: 20°

Обработка ошибок

Если задача в цепочке выбросила исключение, следующие шаги thenApply пропускаются, а exceptionally позволяет подставить запасное значение:

void main() {
    String result = CompletableFuture
            .supplyAsync(() -> Integer.parseInt("сто"))
            .thenApply(n -> "Число: " + n)
            .exceptionally(e -> "Ошибка: " + e.getCause().getMessage())
            .join();
 
    IO.println(result);
}

Результат:

Ошибка: For input string: "сто"

Исключение приходит в exceptionally обёрнутым в CompletionException, поэтому исходную ошибку достают через getCause().

Частая ошибка

Если не дождаться результата, программа может завершиться раньше фоновой задачи. Потоки общего пула — фоновые (daemon): JVM не ждёт их при выходе из main:

void pause(long millis) {
    try {
        Thread.sleep(millis);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}
 
void main() {
    CompletableFuture.runAsync(() -> {
        pause(500);
        IO.println("Отчёт сохранён");
    });
    IO.println("Программа завершена");
}

Результат:

Программа завершена

Строка «Отчёт сохранён» так и не появилась. Сохраните обещание в переменную и вызовите у него join() перед выходом.

Спасибо, что читаете!

Первая асинхронная задача

Цепочка шагов

Объединение двух задач

Много задач

Обработка ошибок

Частая ошибка

Спасибо, что читаете!