thenCompose + benutzerdefinierter Executor + Timeouts

1. thenCompose vs. thenApply: Unterschied und wann was verwenden

In der asynchronen Programmierung mit Java (über CompletableFuture) müssen oft Aktionsketten ausgeführt werden. Dafür gibt es zwei ähnliche Methoden: thenApply und thenCompose. Sie arbeiten jedoch unterschiedlich!

thenApply

Die Methode thenApply wird verwendet, wenn der nächste Schritt eine einfache Werttransformation ist, ohne neue asynchrone Operationen zu starten. Sie erhält das Ergebnis des vorherigen Schritts, verarbeitet es und gibt einen neuen Wert zurück (kein CompletableFuture).

Wenn Sie das Stream API kennen, verhält sich thenApply ungefähr wie map: nimmt das Ergebnis, wendet eine Funktion an und gibt die transformierte Variante zurück.

Beispiel:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "42");
CompletableFuture<Integer> lengthFuture = cf.thenApply(s -> s.length());
// lengthFuture enthält 2 (Länge der Zeichenfolge "42")

Kurz gesagt, thenApply ist eine Möglichkeit zu sagen: „Wenn das Ergebnis bereit ist, mach Folgendes damit“.

thenCompose

• Wird verwendet, wenn der nächste Schritt eine weitere asynchrone Operation ist (gibt ein CompletableFuture zurück).
• Ermöglicht das „Ausklappen“ verschachtelter CompletableFuture (analog zu flatMap).
• Wenn man thenApply mit einer asynchronen Funktion verwendet, erhält man CompletableFuture<CompletableFuture<T>> – unhandlich!

Beispiel:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "user42");

// Angenommen, wir müssen anhand des Benutzernamens seine Bestellungen abrufen (asynchron)
CompletableFuture<List<Order>> ordersFuture = cf.thenCompose(username -> fetchOrdersAsync(username));
// fetchOrdersAsync gibt CompletableFuture<List<Order>> zurück

Anschaulich:

• thenApply: CF<String> → thenApply(s -> s.length()) → CF<Integer>
• thenCompose: CF<User> → thenCompose(u -> fetchOrdersAsync(u.id)) → CF<List<Order>>

Wann was verwenden?

• Die Funktion gibt einen normalen Wert zurück – verwenden Sie thenApply.
• Die Funktion gibt ein CompletableFuture zurück – verwenden Sie thenCompose.

Beispiel für einen Fehler:

cf.thenApply(username -> fetchOrdersAsync(username)); // Sie erhalten CF<CF<List<Order>>>
cf.thenCompose(username -> fetchOrdersAsync(username)); // Sie erhalten CF<List<Order>>

2. Thread-Pool-Steuerung (Executor): Warum und wie man einen eigenen Executor verwendet

Standardmäßig: ForkJoinPool.commonPool()

Wenn Sie CompletableFuture.supplyAsync(...) oder thenApplyAsync(...) ohne Angabe eines Executor schreiben, verwendet Java den gemeinsamen Thread-Pool – ForkJoinPool.commonPool(). Das ist praktisch, passt aber nicht immer:

• Wenn Sie viele lange oder blockierende Operationen haben (Netzwerkaufrufe, Datei-I/O), kann der gemeinsame Pool „verstopfen“ und alle Aufgaben müssen warten.
• Manchmal müssen Aufgaben mit unterschiedlichen Prioritäten isoliert oder die Anzahl gleichzeitig laufender Threads begrenzt werden.

Wann braucht man einen eigenen Executor?

• Lange, blockierende Operationen (z. B. DB-Abfragen, HTTP-Requests, Dateilesen).
• Isolation von Aufgaben: damit Benutzeraufgaben Systemaufgaben nicht beeinträchtigen.
• Ressourcenbegrenzung: z. B. nicht mehr als 10 gleichzeitige Downloads starten.

Wie erstellt man einen eigenen Executor

In der Regel verwendet man ThreadPoolExecutor oder Fabriken aus Executors:

ExecutorService myExecutor = Executors.newFixedThreadPool(10);

Wie verwendet man den eigenen Executor mit CompletableFuture

• Bei den Methoden supplyAsync, runAsync, thenApplyAsync, thenComposeAsync und anderen kann man als zweiten Parameter den eigenen Executor übergeben.

Beispiele:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(
    () -> loadDataFromNetwork(), myExecutor
);

cf.thenApplyAsync(data -> processData(data), myExecutor)
  .thenAcceptAsync(result -> System.out.println(result), myExecutor);

Wichtig: Wenn kein Executor angegeben wird, kommt ForkJoinPool.commonPool() zum Einsatz.

Wann reicht der Standard-Executor?

• Für kurze, CPU-gebundene Aufgaben (einfache Berechnungen).
• Wenn es egal ist, in welchem Thread die Aufgabe ausgeführt wird.

3. Umgang mit Timeouts: orTimeout und completeOnTimeout

Asynchrone Operationen können hängen bleiben oder zu lange dauern (z. B. wenn ein Server nicht antwortet). Um nicht ewig zu warten, bietet CompletableFuture Methoden zum Umgang mit Timeouts.

orTimeout

• Beendet das CompletableFuture mit der Ausnahme TimeoutException, wenn die Operation innerhalb der angegebenen Zeit nicht abgeschlossen wurde.
• Bricht die tatsächlich laufende Aufgabe nicht ab, aber die Downstream-Kette erhält einen Fehler.

Syntax:

cf.orTimeout(3, TimeUnit.SECONDS)
  .exceptionally(ex -> {
      System.out.println("Zeitüberschreitung: " + ex);
      return null;
  });

Beispiel:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(5000); // wir simulieren eine lange Operation
    return "OK";
});

cf.orTimeout(2, TimeUnit.SECONDS)
  .exceptionally(ex -> {
      System.out.println("Fehler: " + ex);
      return "TIMEOUT";
  });

Ergebnis:

Nach 2 Sekunden wird eine TimeoutException ausgelöst, und exceptionally verarbeitet den Fehler.

completeOnTimeout

• Beendet das CompletableFuture mit dem angegebenen Wert, wenn die Operation innerhalb des Timeouts nicht abgeschlossen wurde.
• Löst keine Ausnahme aus, sondern liefert einen „Fallback“-Wert.

Syntax:

cf.completeOnTimeout("DEFAULT", 2, TimeUnit.SECONDS);

Beispiel:

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(5000);
    return "OK";
});

cf.completeOnTimeout("TIMEOUT", 2, TimeUnit.SECONDS)
  .thenAccept(System.out::println); // In 2 Sekunden wird "TIMEOUT" ausgegeben

Vergleich von orTimeout und completeOnTimeout

orTimeout

completeOnTimeout

4. Praxis: Beispiel mit thenCompose, benutzerdefiniertem Executor und Timeout

Aufgabe:

• Einen Benutzer per ID abrufen (asynchron, mit Verzögerung).
• Danach asynchron die Bestellungen des Benutzers abrufen (ebenfalls mit Verzögerung).
• Einen benutzerdefinierten Executor verwenden.
• Einen Timeout für das Abrufen der Bestellungen hinzufügen.

import java.util.concurrent.*;
import java.util.*;

public class AsyncDemo {
    static ExecutorService ioExecutor = Executors.newFixedThreadPool(4);

    // Simulation des asynchronen Abrufens eines Benutzers
    static CompletableFuture<String> fetchUserAsync(int userId) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            return "user" + userId;
        }, ioExecutor);
    }

    // Simulation des asynchronen Abrufens der Bestellungen eines Benutzers
    static CompletableFuture<List<String>> fetchOrdersAsync(String username) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(3000); // Lange Operation!
            return List.of("order1", "order2");
        }, ioExecutor);
    }

    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException ignored) {}
    }

    public static void main(String[] args) {
        fetchUserAsync(42)
            .thenCompose(username ->
                fetchOrdersAsync(username)
                    .orTimeout(2, TimeUnit.SECONDS) // Timeout für das Abrufen der Bestellungen
                    .exceptionally(ex -> {
                        System.out.println("Bestellungen konnten nicht abgerufen werden: " + ex);
                        return List.of();
                    })
            )
            .thenAccept(orders -> System.out.println("Bestellungen: " + orders))
            .join(); // Warten auf den Abschluss der gesamten Kette

        ioExecutor.shutdown();
    }
}

Was passiert:

• Wir holen den Benutzer ab (1 Sekunde).
• Wir holen die Bestellungen ab (3 Sekunden, aber Timeout 2 Sekunden).
• Wenn es nicht rechtzeitig gelingt – wir fangen die TimeoutException und geben eine leere Liste zurück.
• Alles läuft auf dem benutzerdefinierten Executor.

Ergebnis:

Bestellungen konnten nicht abgerufen werden: java.util.concurrent.TimeoutException
Bestellungen: []

Wenn Sie die Verzögerung in fetchOrdersAsync auf 1_000 ms verringern – sehen Sie die echten Bestellungen.

5. Häufige Fehler und Besonderheiten

Fehler Nr. 1: Verwendung von thenApply statt thenCompose für asynchrone Operationen.
Wenn eine Funktion ein CompletableFuture zurückgibt und Sie thenApply anwenden, erhalten Sie den verschachtelten Typ CompletableFuture<CompletableFuture<T>>. Das verkompliziert die Kette und führt zu unnötigen Hüllen. Lösung: Verwenden Sie thenCompose, um das Ergebnis in CompletableFuture<T> zu „entpacken“.

Fehler Nr. 2: Starten von langen oder I/O-Aufgaben ohne eigenen Executor.
Standardmäßig werden Aufgaben im ForkJoinPool.commonPool() ausgeführt. Wenn dieser überlastet wird, steigen die Latenzen und andere Aufgaben der Anwendung können langsamer werden. Lösung: Erstellen Sie einen eigenen ExecutorService und übergeben Sie ihn an supplyAsync/thenApplyAsync.

Fehler Nr. 3: Die Erwartung, dass orTimeout die Aufgabe abbricht.
orTimeout beendet lediglich das CompletableFuture mit einer Ausnahme aufgrund eines Timeouts, aber die Aufgabe selbst läuft im Hintergrund weiter. Lösung: Wenn die Ausführung gestoppt werden soll, verwenden Sie cancel(true) oder eigene Unterbrechungsmechanismen.

Fehler Nr. 4: Falsches Verständnis des Geltungsbereichs des Timeouts.
orTimeout und completeOnTimeout wirken jeweils nur auf einen konkreten Schritt der Kette, nicht auf die gesamte Kette. Lösung: Wenn ein globales Timeout für die gesamte Kette benötigt wird, kapseln Sie sie in ein eigenes CompletableFuture und wenden Sie das Timeout darauf an.

Fehler Nr. 5: Kein Beenden des ExecutorService.
Wenn nach Abschluss der Aufgaben nicht shutdown()/shutdownNow() auf dem ExecutorService aufgerufen wird, laufen Threads weiter und das Programm kann „hängen bleiben“. Lösung: Beenden Sie den ExecutorService immer in finally oder verwenden Sie try-with-resources in Java 21+.

1

Umfrage/Quiz

Asynchrones Programmieren, Level 55, Lektion 4

Nicht verfügbar

Asynchrones Programmieren

Asynchrones Programmieren

Öffnen