Java Structured Concurrency | W3Docs Learn Java

Structured Concurrency behandelt eine Gruppe von nebenläufigen Teilaufgaben als eine einzelne Arbeitseinheit: Sie werden gemeinsam gestartet, gemeinsam beendet, und wenn eine fehlschlägt oder der Aufrufer abgebrochen wird, werden die übrigen ebenfalls abgebrochen — keine verwaisten Threads, die länger leben als der Block, der sie gestartet hat. Das Modell wird durch java.util.concurrent.StructuredTaskScope bereitgestellt (eine Preview-API, die in Java 21 eingeführt wurde) und basiert auf denselben Virtual Threads, die weiter vorne in diesem Teil behandelt werden. Das Ziel ist einfach: nebenläufigen Code so leicht lesbar, debuggbar und nachvollziehbar zu machen wie eine einfache sequenzielle Methode.

Dieses Kapitel erklärt, warum „structured" wichtig ist, den Aufbau eines Task-Scopes, die zwei eingebauten Shutdown-Richtlinien, wie Deadlines und Abbrüche weitergegeben werden, und ein ausführbares Arbeitsbeispiel. Es setzt voraus, dass Sie mit dem Executor-Framework und Callable/Future vertraut sind.

Warum „structured"?

Klassische Thread-Pools sind unstrukturiert: Sie submitten eine Aufgabe an einen gemeinsam genutzten ExecutorService und erhalten ein Future zurück, dessen Lebensdauer nichts mit der Methode zu tun hat, die es erstellt hat. Eine Aufgabe kann ihren Aufrufer überleben, ein Fehler in einer Aufgabe ist für ihre Geschwister unsichtbar, und Abbrüche müssen manuell verdrahtet werden. Das Ergebnis sind verlorene Threads und verschachteltes Fehler-Handling.

Structured Concurrency übernimmt die Disziplin des strukturierten Kontrollflusses: So wie ein try-Block seine Anweisungen abgrenzt, begrenzt ein Task-Scope seine Teilaufgaben. Innerhalb eines Blocks verzweigte Teilaufgaben müssen alle abgeschlossen sein, bevor der Block endet. Lebenszeiten verschachteln sich sauber, sodass ein Thread-Dump und ein Stack-Trace tatsächlich zeigen, wer was gestartet hat.

Aspekt	Unstrukturiert (`ExecutorService` gemeinsamer Pool)	Strukturiert (`StructuredTaskScope`)
Lebensdauer der Teilaufgabe	Unabhängig vom Aufrufer	Begrenzt durch den umschließenden Block
Fehler in einer Teilaufgabe	Versteckt in einem `Future` bis zum Aufruf von `get`	Kann den gesamten Scope kurzschließen
Abbruch	Manuell, leicht zu vergessen	Automatisch bei Fehler oder Interrupt
Ressourcenbereinigung	Liegt bei Ihnen	`close()` wartet auf jede Teilaufgabe

Der Aufbau eines Scopes

Ein Scope ist ein AutoCloseable und lebt daher in einem try-with-resources-Block. Sie forken Teilaufgaben (jede gibt ein Subtask-Handle zurück), rufen join() auf, um auf sie zu warten, und lesen dann jedes Ergebnis. Die ShutdownOnFailure-Richtlinie bricht verbleibende Teilaufgaben ab, sobald eine von ihnen eine Ausnahme wirft:

import java.util.concurrent.StructuredTaskScope;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    StructuredTaskScope.Subtask<String> user  = scope.fork(() -> fetchUser(id));
    StructuredTaskScope.Subtask<Integer> order = scope.fork(() -> fetchOrderCount(id));

    scope.join();            // wait for both branches
    scope.throwIfFailed();   // rethrow if either branch failed

    return new Profile(user.get(), order.get());
}   // close() guarantees both subtasks have ended before we leave

Wenn fetchUser eine Ausnahme wirft, unterbricht ShutdownOnFailure das noch laufende fetchOrderCount, join() kehrt zurück, und throwIfFailed() wirft die ursprüngliche Ursache verpackt in einer ExecutionException erneut. Es geht kein Thread verloren.

Eingebaute Shutdown-Richtlinien

Die zwei mitgelieferten Richtlinien decken die gängigen Muster ab; für alles andere leiten Sie von StructuredTaskScope ab.

Richtlinie	Endet wenn	Verwenden Sie sie für
`ShutdownOnFailure`	Alle erfolgreich, oder eine schlägt fehl	Fan-out, wenn Sie jedes Ergebnis benötigen (der häufige Fall)
`ShutdownOnSuccess<T>`	Erste Erfolgsmeldung, oder alle schlagen fehl	Redundante Quellen gegeneinander antreten lassen; die schnellste Antwort nehmen

ShutdownOnSuccess gibt den Gewinner über result() zurück und bricht die Verlierer ab:

try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    scope.fork(() -> queryMirrorA());
    scope.fork(() -> queryMirrorB());
    scope.join();
    return scope.result();   // the first one to return; the slower is cancelled
}

Deadlines und Abbrüche werden weitergegeben

Ein Scope kann mit einer Deadline gewartet werden; wenn diese abläuft, werden unfertige Teilaufgaben abgebrochen:

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    scope.fork(() -> slowService());
    scope.joinUntil(Instant.now().plusSeconds(2));  // throws TimeoutException if late
    scope.throwIfFailed();
}

Der Abbruch ist kooperativ und fließt nach unten: Wenn der Thread, dem der Scope gehört, unterbrochen wird, werden alle Teilaufgaben der Reihe nach unterbrochen. Da jede Teilaufgabe auf ihrem eigenen Virtual Thread läuft, ist das Erstellen von Tausenden von ihnen günstig — der Scope, nicht eine feste Poolgröße, ist die Einheit, über die Sie nachdenken.

Ein Arbeitsbeispiel: Fan-out, Fehler und das Joinen einer Liste

StructuredTaskScope ist ein Preview-Feature. Um dieses Beispiel auf einem stabilen JDK ausführbar zu halten, modellieren wir dieselbe Idee mit einem Virtual-Thread-per-task-Executor: ein try-with-resources-Block, der eine Gruppe von Teilaufgaben abgrenzt und erst endet, wenn jeder Teilaufgaben-Thread beendet ist. Er fächert zwei Aufrufe gleichzeitig auf, zeigt dann, wie ein Fehler die Arbeitseinheit kurzschließt, und wie invokeAll eine ganze Liste auf einmal joinen kann.

java— editable, runs on the server

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class StructuredConcurrencyDemo {

// Simulate fetching a piece of data over the network.
  static String fetch(String name, long millis) throws InterruptedException {
    Thread.sleep(millis);
    return name + "@" + Thread.currentThread();
  }

public static void main(String[] args) throws Exception {
    // A virtual-thread-per-task executor scopes a group of subtasks to one block.
    // try-with-resources guarantees the scope shuts down before we move on:
    // the SAME structured-concurrency idea StructuredTaskScope formalizes.
    System.out.println("=== happy path: fan out, join, combine ===");
    try (ExecutorService scope = Executors.newVirtualThreadPerTaskExecutor()) {
      long start = System.currentTimeMillis();

Future<String> user  = scope.submit(() -> fetch("user", 120));
      Future<String> order = scope.submit(() -> fetch("order", 200));

// get() joins each subtask; the block does not finish until both return.
      String u = user.get();
      String o = order.get();

long elapsed = System.currentTimeMillis() - start;
      System.out.println("user thread is virtual : " + u.contains("VirtualThread"));
      System.out.println("order thread is virtual: " + o.contains("VirtualThread"));
      // 120ms and 200ms ran concurrently, so wall time is ~max, not the sum.
      System.out.println("ran concurrently (<320ms): " + (elapsed < 320) + " (" + elapsed + "ms)");
    } // scope.close() blocks until every subtask thread has terminated

System.out.println();
    System.out.println("=== failure short-circuits the unit of work ===");
    try (ExecutorService scope = Executors.newVirtualThreadPerTaskExecutor()) {
      Future<String> good = scope.submit(() -> fetch("good", 300));
      Future<String> bad  = scope.submit(() -> {
        Thread.sleep(40);
        throw new IllegalStateException("upstream said no");
      });

try {
        // Join the failing branch first: it throws at 40ms while good
        // is still sleeping (300ms), so the sibling is genuinely running.
        bad.get(); // joining the failed subtask surfaces its exception
        good.get();
      } catch (ExecutionException e) {
        // The cause is the real error thrown inside the subtask.
        System.out.println("caught: " + e.getCause().getClass().getSimpleName()
            + " -> " + e.getCause().getMessage());
        // We cancel the sibling so no orphan keeps running past the block.
        good.cancel(true);
        System.out.println("sibling cancelled: " + good.isCancelled());
      }
    }

System.out.println();
    System.out.println("=== invokeAll: join a whole list as one operation ===");
    try (ExecutorService scope = Executors.newVirtualThreadPerTaskExecutor()) {
      List<Callable<Integer>> tasks = List.of(
          () -> { Thread.sleep(60); return 10; },
          () -> { Thread.sleep(30); return 20; },
          () -> { Thread.sleep(90); return 30; });

// invokeAll returns only when ALL tasks are done (or the timeout hits).
      List<Future<Integer>> results = scope.invokeAll(tasks, 2, TimeUnit.SECONDS);
      int sum = 0;
      for (Future<Integer> f : results) sum += f.get();
      System.out.println("all " + results.size() + " joined, sum = " + sum);
    }

System.out.println();
    System.out.println("done; all scopes closed");
  }
}

Was aus der Ausführung zu entnehmen ist:

Beide Teilaufgaben meldeten is virtual : true — jedes submit lief auf seinem eigenen Virtual Thread, demselben leichtgewichtigen Träger, den StructuredTaskScope.fork verwendet, sodass das Erstellen eines Threads pro Teilaufgabe günstig ist.
Der Happy-Path-Block gab ran concurrently (<320ms): true aus, obwohl die beiden Fetches 120ms und 200ms schlafen: Sie überlappten sich, sodass die Wandzeit dem langsamsten Zweig (~200ms) folgt, nicht der Summe (320ms). Diese Überlappung ist der eigentliche Zweck des Fan-outs.
Das Verlassen des try-with-resources-Blocks rief close() auf, was blockierte, bis jeder Teilaufgaben-Thread beendet war — der Scope ist die Lebenszeit-Einheit, genau die Disziplin, die StructuredTaskScope konstruktionsbedingt erzwingt.
Im Fehlerabschnitt gab das Programm caught: IllegalStateException -> upstream said no aus: ein innerhalb einer Teilaufgabe geworfener Fehler erscheint am Join-Punkt verpackt in ExecutionException, und getCause() gibt Ihnen die ursprüngliche Ausnahme zurück.
Nach dem Abfangen des Fehlers wurde sibling cancelled: true ausgegeben — wir haben den noch laufenden good-Zweig abgebrochen, damit kein Waisenkind den Block überlebt, was genau das ist, was ShutdownOnFailure automatisch für Sie erledigt; hier haben wir es manuell getan, um den Mechanismus zu zeigen.

Übungen

Übung

Was passiert mit den anderen geforkte Teilaufgaben bei StructuredTaskScope.ShutdownOnFailure, wenn eine von ihnen eine Ausnahme wirft?

Die verbleibenden Teilaufgaben werden abgebrochen, und join() gefolgt von throwIfFailed() wirft den ursprünglichen Fehler verpackt in einer ExecutionException erneutDie anderen Teilaufgaben laufen unabhängig weiter und Sie müssen jede selbst abbrechen, nachdem der Block endetDie gesamte JVM wird sofort beendet, weil eine nicht abgefangene Ausnahme einem Thread entkommen istEs passiert nichts, bis Sie jedes Subtask-Handle abfragen, da Fehler stillschweigend gespeichert werden wie in einem normalen Future