CompletableFuture: asynchron ausführen und Ergebnisse kombinieren

1. Überblick

Eine API muss oft mehrere Services aufrufen: Kundendaten abrufen, Bestellungen, Versandkosten. Werden diese nacheinander aufgerufen, entspricht die Gesamtzeit der Summe aller einzelnen Latenzen. CompletableFuture ermöglicht es, unabhängige Aufrufe parallel auszuführen und die Ergebnisse danach zu kombinieren.

Nach diesem Artikel werden Sie wissen, wie Sie asynchrone Aufgaben erstellen, Verarbeitungsschritte verketten (thenApply, thenCompose) und Ergebnisse kombinieren (thenCombine, allOf, anyOf). Sie werden auch lernen, wie Sie Fehler behandeln, Timeouts setzen und Tests für asynchronen Code schreiben.

2. Vorbereitung des Beispiels

Wir verwenden drei Mock-Services. Jeder Service “schläft” eine Weile, um Netzwerkverzögerungen zu simulieren:

import java.math.BigDecimal;
import java.util.List;

record Customer(long id, String name, String city) {}
record Order(long id, long customerId, BigDecimal total) {}

final class Delay {
    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }
}

class CustomerService {
    Customer findById(long id) {
        Delay.sleep(300);
        return new Customer(id, "Lan", "Hue");
    }
}

class OrderService {
    List<Order> findByCustomer(long customerId) {
        Delay.sleep(400);
        return List.of(
            new Order(1, customerId, new BigDecimal("150000")),
            new Order(2, customerId, new BigDecimal("90000")));
    }
}

class ShippingService {
    BigDecimal feeFor(String city) {
        Delay.sleep(200);
        return "Hue".equals(city) ? new BigDecimal("30000") : new BigDecimal("45000");
    }
}

Diese drei Services verwenden nur die JDK. Sie benötigen keine zusätzlichen Bibliotheken zu ihrer Ausführung.

3. Asynchrone Aufgaben erstellen

3.1. supplyAsync und runAsync

supplyAsync führt einen Supplier aus und gibt CompletableFuture<T> zurück. runAsync führt einen Runnable ohne Ergebnis aus und gibt CompletableFuture<Void> zurück:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CreateDemo {
    public static void main(String[] args) {
        var customerService = new CustomerService();

        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            CompletableFuture<Customer> customerFuture =
                CompletableFuture.supplyAsync(() -> customerService.findById(42L), executor);

            CompletableFuture<Void> auditFuture =
                CompletableFuture.runAsync(() -> System.out.println("Zugriff protokollieren"), executor);

            System.out.println("Main-Thread wird nicht blockiert");
            System.out.println(customerFuture.join());
            auditFuture.join();
        }
    }
}

Main-Thread wird nicht blockiert
Zugriff protokollieren
Customer[id=42, name=Lan, city=Hue]

Die Reihenfolge der ersten beiden Zeilen kann variieren, da sie auf zwei unterschiedlichen Threads laufen. Sie sollten folgende Punkte beachten:

• Der zweite Parameter ist ein Executor. Falls weggelassen, läuft die Aufgabe auf ForkJoinPool.commonPool().
• Executors.newVirtualThreadPerTaskExecutor() (Java 21) erstellt einen virtual thread für jede Aufgabe. Diese Wahl ist ideal für I/O-Aufgaben wie HTTP-Aufrufe oder Datenbankabfragen.
• ExecutorService ist seit Java 19 AutoCloseable. Der try-with-resources-Block wartet auf den Abschluss aller Aufgaben, bevor der Executor geschlossen wird.
• join() blockiert den aktuellen Thread, bis ein Ergebnis verfügbar ist. Falls die Aufgabe fehlschlägt, wirft join() CompletionException (unchecked). get() wirft dagegen ExecutionException (checked).

3.2. Ein Future manuell abschließen

Manchmal kommt das Ergebnis von einem Callback statt von einem Lambda. In diesem Fall erstellen Sie ein leeres Future und schließen es manuell ab:

CompletableFuture<String> paymentResult = new CompletableFuture<>();

// Irgendwo, z.B. im Callback eines Payment-Gateways:
paymentResult.complete("PAID");

System.out.println(paymentResult.join()); // PAID

Um einen Fehler zu melden, rufen Sie completeExceptionally(new RuntimeException("Karte abgelehnt")) auf. Wenn Sie bereits einen Wert haben, ist CompletableFuture.completedFuture(value) die kürzeste Möglichkeit, ein bereits abgeschlossenes Future zu erstellen.

4. Verarbeitungsschritte verketten

4.1. thenApply und thenAccept

thenApply transformiert das Ergebnis, ähnlich wie map bei Stream. thenAccept konsumiert das Ergebnis und gibt nichts zurück:

CompletableFuture<String> greeting = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenApply(Customer::name)
    .thenApply(name -> "Hallo, " + name);

greeting.thenAccept(System.out::println).join();

Hallo, Lan

Jeder Aufruf von thenApply gibt ein neues CompletableFuture zurück. Das ursprüngliche Future wird nicht verändert.

4.2. thenCompose, wenn der nächste Schritt auch asynchron ist

Um Bestellungen abzurufen, benötigen Sie die customerId aus dem vorherigen Schritt. Wenn Sie thenApply mit einer Funktion verwenden, die ein Future zurückgibt, wird das Ergebnis CompletableFuture<CompletableFuture<List<Order>>>. thenCompose “flacht” diesen verschachtelten Typ ab, ähnlich wie flatMap:

CompletableFuture<List<Order>> ordersFuture = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenCompose(customer -> CompletableFuture.supplyAsync(
        () -> orderService.findByCustomer(customer.id()), executor));

System.out.println(ordersFuture.join().size()); // 2

Diese zwei Schritte hängen voneinander ab, daher laufen sie noch sequenziell. Die Gesamtzeit liegt bei etwa 700 ms. thenCompose beschleunigt nicht. Es hilft Ihnen, Schritte zu verketten, ohne einen Thread zu blockieren.

5. Ergebnisse kombinieren

Mit unabhängigen Aufgaben zeigt CompletableFuture seine wahre Stärke.

5.1. thenCombine für zwei Futures

Angenommen, wir kennen bereits die Stadt des Kunden und möchten gleichzeitig die Bestellungsliste und die Versandkosten abrufen:

long start = System.currentTimeMillis();

CompletableFuture<List<Order>> orders =
    CompletableFuture.supplyAsync(() -> orderService.findByCustomer(42L), executor);
CompletableFuture<BigDecimal> shippingFee =
    CompletableFuture.supplyAsync(() -> shippingService.feeFor("Hue"), executor);

CompletableFuture<BigDecimal> grandTotal = orders.thenCombine(shippingFee,
    (orderList, fee) -> orderList.stream()
        .map(Order::total)
        .reduce(BigDecimal.ZERO, BigDecimal::add)
        .add(fee));

System.out.println(grandTotal.join());
System.out.println("Dauerte " + (System.currentTimeMillis() - start) + " ms");

270000
Dauerte 405 ms

Die zwei Aufrufe laufen parallel. Die Gesamtzeit entspricht ungefähr dem langsamsten Aufruf (400 ms), statt 600 ms bei sequenzieller Ausführung. Die Funktion im zweiten Parameter von thenCombine wird nur ausgeführt, wenn beide Futures abgeschlossen sind.

5.2. allOf für mehrere Futures

Wenn Sie mehr als zwei Aufgaben haben, verwenden Sie allOf. Es gibt CompletableFuture<Void> zurück, daher müssen Sie die Ergebnisse selbst aus jedem Sub-Future extrahieren:

List<String> cities = List.of("Hue", "Hanoi", "Danang");

List<CompletableFuture<BigDecimal>> fees = cities.stream()
    .map(city -> CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor))
    .toList();

CompletableFuture<List<BigDecimal>> allFees = CompletableFuture
    .allOf(fees.toArray(CompletableFuture[]::new))
    .thenApply(ignored -> fees.stream()
        .map(CompletableFuture::join)
        .toList());

System.out.println(allFees.join());

[30000, 45000, 45000]

Innerhalb von thenApply sind alle Sub-Futures abgeschlossen. Daher gibt join() sofort zurück, ohne zu blockieren. Die Reihenfolge der Ergebnisse entspricht der Reihenfolge in cities.

5.3. anyOf – das schnellste Ergebnis

anyOf wird abgeschlossen, sobald ein Sub-Future abgeschlossen ist. Dies ist nützlich, wenn Sie mehrere Replicas desselben Service aufrufen:

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> { Delay.sleep(500); return "replica-A"; }, executor),
    CompletableFuture.supplyAsync(() -> { Delay.sleep(100); return "replica-B"; }, executor));

System.out.println(fastest.join()); // replica-B

Der Rückgabetyp ist CompletableFuture<Object>, daher müssen Sie selbst casten. Beachten Sie: Wenn das schnellste Future mit einem Fehler endet, endet anyOf auch mit diesem Fehler.

6. Fehlerbehandlung und Timeouts

6.1. exceptionally und handle

Eine Exception in einem beliebigen Schritt wird an die nachfolgenden Schritte weitergegeben. Diese Schritte werden übersprungen, bis auf eine Fehlerbehandlung:

CompletableFuture<BigDecimal> fee = CompletableFuture
    .supplyAsync(() -> {
        throw new IllegalStateException("Shipping Service ist down");
    }, executor)
    .thenApply(value -> (BigDecimal) value)          // wird übersprungen
    .exceptionally(ex -> {
        System.out.println("Fehler: " + ex.getMessage());
        return new BigDecimal("50000");              // Fallback-Wert
    });

System.out.println(fee.join());

Fehler: java.lang.IllegalStateException: Shipping Service ist down
50000

Der Parameter ex ist normalerweise eine CompletionException, die die ursprüngliche Exception einschließt. Daher hat die Nachricht einen Klassennamen-Präfix. Verwenden Sie ex.getCause(), um den echten Fehler zu erhalten.

handle empfängt sowohl Ergebnis als auch Fehler. Es wird immer aufgerufen, unabhängig davon, ob die Aufgabe erfolgreich ist oder fehlschlägt:

CompletableFuture<String> status = CompletableFuture
    .supplyAsync(() -> orderService.findByCustomer(42L), executor)
    .handle((orderList, ex) -> ex == null
        ? "Es gibt " + orderList.size() + " Bestellungen"
        : "Bestellungen konnten nicht geladen werden");

Zusätzlich gibt es noch whenComplete. Es empfängt auch Ergebnis und Fehler, ändert aber das Ergebnis nicht. Diese Methode eignet sich zum Protokollieren.

6.2. orTimeout und completeOnTimeout

Sie sollten niemals auf einen externen Service unbegrenzt warten. Java 9+ bietet zwei Methoden dafür:

import java.util.concurrent.TimeUnit;

CompletableFuture<BigDecimal> withDefault = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .completeOnTimeout(new BigDecimal("45000"), 500, TimeUnit.MILLISECONDS);

System.out.println(withDefault.join()); // 45000 nach ~500 ms

CompletableFuture<BigDecimal> strict = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .orTimeout(500, TimeUnit.MILLISECONDS);
// strict.join() wirft CompletionException mit TimeoutException als Ursache

completeOnTimeout gibt einen Standardwert zurück, wenn die Zeit abläuft. orTimeout schließt das Future mit TimeoutException ab.

Hinweis: Der Timeout schließt nur das Future ab. Er stoppt die Aufgabe nicht. Der Thread schläft weiterhin 2 Sekunden. cancel(true) unterbricht auch nicht den laufenden Thread. Wenn Sie die Aufgabe wirklich abbrechen müssen, müssen Sie ein Abbruch-Flag in der Aufgabe überprüfen oder einen Timeout im HTTP-Client selbst setzen.

7. Tests schreiben

Wenn Sie Spring Boot verwenden, hat spring-boot-starter-test bereits JUnit 5 und AssertJ. Bei reinen Java-Projekten fügen Sie folgende Abhängigkeiten hinzu:

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.assertj</groupId>
    <artifactId>assertj-core</artifactId>
    <version>3.26.3</version>
    <scope>test</scope>
</dependency>

Zunächst extrahieren Sie die Logik in eine Klasse, damit sie leicht zu testen ist:

class CheckoutService {
    private final OrderService orderService;
    private final ShippingService shippingService;
    private final Executor executor;

    CheckoutService(OrderService o, ShippingService s, Executor executor) {
        this.orderService = o;
        this.shippingService = s;
        this.executor = executor;
    }

    CompletableFuture<BigDecimal> grandTotal(long customerId, String city) {
        var orders = CompletableFuture.supplyAsync(() -> orderService.findByCustomer(customerId), executor);
        var fee = CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor);
        return orders.thenCombine(fee, (list, f) -> list.stream()
            .map(Order::total)
            .reduce(BigDecimal.ZERO, BigDecimal::add)
            .add(f));
    }
}

Dann schreiben Sie Tests sowohl für erfolgreiche als auch für fehlgeschlagene Fälle:

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import org.junit.jupiter.api.Test;

class CheckoutServiceTest {

    @Test
    void shouldCombineOrdersAndShippingFee() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), new ShippingService(), executor);

            assertThat(service.grandTotal(42L, "Hue"))
                .succeedsWithin(Duration.ofSeconds(2))
                .isEqualTo(new BigDecimal("270000"));
        }
    }

    @Test
    void shouldPropagateShippingFailure() {
        ShippingService broken = new ShippingService() {
            @Override
            BigDecimal feeFor(String city) {
                throw new IllegalStateException("down");
            }
        };

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), broken, executor);

            assertThatThrownBy(() -> service.grandTotal(42L, "Hue").join())
                .isInstanceOf(CompletionException.class)
                .hasCauseInstanceOf(IllegalStateException.class);
        }
    }
}

succeedsWithin wartet maximal 2 Sekunden und überprüft dann den Wert. Dadurch bleibt der Test nicht hängen, falls der Code ein Problem hat. Im zweiten Test wird der Fehler von ShippingService durch thenCombine weitergegeben und in CompletionException eingewickelt.

In Tests können Sie Runnable::run als Executor übergeben, damit alles synchron auf dem Test-Thread läuft. Dies macht Tests schneller und die Ergebnisse stabiler.

8. Häufige Fehler

• I/O auf dem commonPool ausführen. Dieser Pool hat so viele Threads wie CPUs minus eins und wird von der ganzen Anwendung gemeinsam genutzt (auch von parallelen Streams). Ein paar langsame HTTP-Aufrufe können das Ganze blockieren. Bei I/O sollten Sie einen dedizierten Executor übergeben, idealerweise mit virtual threads.
• join() zu früh aufrufen. Wenn Sie supplyAsync(a).join() schreiben und dann supplyAsync(b).join(), ist das sequenziell. Sie müssen erst alle Futures starten, dann kombinieren oder warten.
• Exceptions schlucken. Wenn niemand join(), get() aufruft oder exceptionally/whenComplete hinzufügt, verschwindet der Fehler spurlos.
• thenApply mit thenApplyAsync verwechseln. thenApply kann auf dem Thread des vorherigen Schritts oder auf dem aufrufenden Thread laufen, wenn das Future bereits fertig ist. Falls der nächste Schritt schwer ist oder blockiert, verwenden Sie die ...Async-Variante mit einem Executor.
• Kein Timeout setzen. Ein hängender Service zieht Ihren Request mit sich.
• Programm endet zu früh. Threads im commonPool sind Daemon-Threads. Wenn main endet, bevor die Aufgaben fertig sind, werden die Ergebnisse nie gedruckt.

Hinweis: Sie sollten CompletableFuture nicht verwenden, wenn Sie nur einen einzigen Aufruf haben und sofort auf das Ergebnis warten. Ab Java 21 mit Code, der auf virtual threads läuft, ist synchroner Code oft leichter zu lesen und hat keine Performanz-Einbußen. CompletableFuture ist sinnvoll, wenn Sie mehrere unabhängige Aufgaben parallel ausführen oder eine nicht-blockierende Pipeline benötigen.

9. Fazit

CompletableFuture ermöglicht es Ihnen, unabhängige Aufgaben parallel auszuführen und Ergebnisse zu kombinieren, ohne Threads manuell zu verwalten. Schnelle Zusammenfassung:

• Aufgaben erstellen: supplyAsync/runAsync, immer einen passenden Executor übergeben.
• Schritte verketten: thenApply zum Transformieren von Werten, thenCompose, wenn der nächste Schritt ein Future zurückgibt.
• Kombinieren: thenCombine für zwei Futures, allOf für mehrere, anyOf für das schnellste Ergebnis.
• Fehler und Timeouts: exceptionally, handle, orTimeout, completeOnTimeout.

In Tests verwenden Sie AssertJ’s succeedsWithin, um Ergebnisse zu überprüfen, ohne dass der Test hängbleiben kann.