Structured Concurrency

1. Wprowadzenie

Kiedy wątki grają każdy swoją partię

Zwykła wielowątkowość często przypomina próbę bez dyrygenta. Każdy wątek — jak muzyk, który gra swoją melodię, nie słuchając pozostałych. Ktoś skończył wcześniej i poszedł na papierosa, ktoś utknął na jednym akordzie, ktoś w ogóle pomylił nuty i wyrzucił błąd. W efekcie wychodzi nie symfonia, lecz kakofonia: trudno dociec, kto i gdzie się pomylił, a zatrzymanie wszystkich naraz to nie lada wyzwanie.

Structured Concurrency rozwiązuje ten problem. Zamienia rozproszone wątki w prawdziwy zespół: wszystkie zadania są zebrane pod jednym „dyrygentem”. Jeśli daje sygnał do przerwania — orkiestra milknie. Jeśli jeden muzyk nawalił — pozostali grzecznie się zatrzymują, nie psując ogólnej harmonii. Wszystkie wyniki i błędy zbierane są centralnie, a nie rozrzucone po kątach kodu.

Wyobraź sobie: nie wypuszczasz muzyków, by grali każdy co chce, tylko zbierasz ich w jednej sali. Jest dyrygent, jest partytura i nawet jeśli trąbka fałszuje — orkiestra się nie rozsypuje, lecz elegancko kończy występ.

Co daje Structured Concurrency

• Jednolity „scope” zadania: wszystkie podzadania żyją w ramach jednego bloku kodu, a ich cykl życia jest ograniczony tym blokiem.
• Przewidywalne zakończenie: wątek nadrzędny nie zakończy się, dopóki nie zakończą się wszystkie podzadania.
• Scentralizowane anulowanie: jeśli jedno zadanie się nie powiodło lub zadanie nadrzędne kończy pracę — wszystkie podzadania są poprawnie anulowane.
• Spójna obsługa błędów: błędy podzadań są agregowane, można otrzymać „drzewo przyczyn” (tree of causes).
• Czysty i czytelny kod: brak „wiszących” wątków, brak zapomnianych zadań, brak wyścigów o anulowanie.

Structured Concurrency to nie tylko nowe API, ale nowy sposób myślenia: zadania powinny być strukturyzowane tak jak zwykłe bloki kodu (np. try-with-resources).

2. Status Structured Concurrency w Javie

W chwili pisania tego kursu Structured Concurrency ma status Preview (Java 21–23), ale oczekuje się przejścia do GA (General Availability) w Javie 24/25. API znajduje się w pakiecie jdk.incubator.concurrent. Przed użyciem w produkcji koniecznie sprawdź aktualne release notes swojej wersji JDK!

Najważniejsze klasy:

• StructuredTaskScope — podstawowa klasa do zarządzania grupą zadań.
• Warianty: StructuredTaskScope.ShutdownOnFailure, StructuredTaskScope.ShutdownOnSuccess — polityki zakończenia zadań.

Główne koncepcje StructuredTaskScope

Model: fork, join i przyjazny przegląd wyników

Gdy dyrygent (czyli zadanie nadrzędne) daje znak — podzadania rozchodzą się do swoich partii. Ten moment nazywa się fork — jakbyś wypuszczał muzyków, by grali swoje fragmenty w różnych salach.

Potem przychodzi czas na join — dyrygent podnosi batutę i wszyscy wracają, by zagrać finałowy akord razem.

A dalej można zapytać każdego uczestnika, jak poszło:

• przez resultNow() pobrać wynik od razu, jeśli wszystko zagrano bez błędów;
• przez throwIfFailed() — upewnić się, że nikt nie sfal­szował. Jeśli ktoś jednak pogubił się w nutach, wyrzucane jest jedno wyjąt­kowe zdarzenie — jakby dyrygent powiedział: „Mamy awarię w orkiestrze, zaczynamy od nowa”.

Polityki zakończenia

Każdy dyrygent ma swoje zasady, kiedy zatrzymać muzykę. W Structured Concurrency określa to polityka zakończenia:

• ShutdownOnFailure — jeśli choć jeden muzyk wypadnie z rytmu, dyrygent macha ręką: „Stop! Zaczynamy od początku”. Wszyscy pozostali natychmiast przestają grać.
• ShutdownOnSuccess — przeciwnie, gdy tylko ktoś idealnie zagra swoją partię, dyrygent jest zadowolony: „Wystarczy, dalej nie trzeba, mamy już zwycięzcę”. Pozostali milkną — polityka pierwszej udanej odpowiedzi.

Praca z wirtualnymi wątkami

Każde podzadanie StructuredTaskScope uruchamia w wirtualnym wątku. To tak, jakbyś miał orkiestrę, w której każdy muzyk jest pojętny i szybki, bez kaprysów i wymagań co do sceny. Możesz śmiało tworzyć setki, tysiące takich wykonawców — to nie ciężkie wątki systemowe, lecz niemal bezciężarowe nuty, które wybrzmiewają dokładnie wtedy, gdy trzeba.

3. Przykład: agregator żądań HTTP

Rozważmy praktyczne zadanie: mamy trzy źródła danych (np. trzy różne serwery) i chcemy otrzymać odpowiedź albo od wszystkich (i ją zagregować), albo od pierwszego, który odpowie pomyślnie.

Wariant 1: „Wszystkie muszą się powieść” (ShutdownOnFailure)

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

public class AggregatorAllSuccess {
    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Future<String> f1 = scope.fork(() -> fetchFromSource1());
            Future<String> f2 = scope.fork(() -> fetchFromSource2());
            Future<String> f3 = scope.fork(() -> fetchFromSource3());

            scope.join(); // czekamy na zakończenie wszystkich zadań
            scope.throwIfFailed(); // jeśli choć jedno się wywaliło — wyrzucamy wyjątek

            // Wszystkie zadania zakończyły się powodzeniem — można agregować wyniki
            String result = f1.resultNow() + f2.resultNow() + f3.resultNow();
            System.out.println("Zagregowany wynik: " + result);
        }
    }

    static String fetchFromSource1() { /* ... */ return "A"; }
    static String fetchFromSource2() { /* ... */ return "B"; }
    static String fetchFromSource3() { /* ... */ return "C"; }
}

Co się dzieje:

• Wszystkie trzy zadania uruchamiają się równolegle (w wirtualnych wątkach).
• Jeśli choć jedno się nie powiedzie — pozostałe są anulowane, a wyjątek zostaje rzucony.
• Jeśli wszystkie się powiodą — można bezpiecznie zagregować wyniki.

Wariant 2: „Sukces po pierwszym prawidłowym” (ShutdownOnSuccess)

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

public class AggregatorFirstSuccess {
    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
            Future<String> f1 = scope.fork(() -> fetchFromSource1());
            Future<String> f2 = scope.fork(() -> fetchFromSource2());
            Future<String> f3 = scope.fork(() -> fetchFromSource3());

            scope.join(); // czekamy na pierwszy udany
            scope.throwIfFailed(); // jeśli wszystkie padły — wyrzucamy wyjątek

            String result = scope.result(); // wynik pierwszego udanego zadania
            System.out.println("Pierwszy pomyślny wynik: " + result);
        }
    }

    static String fetchFromSource1() { /* ... */ return "A"; }
    static String fetchFromSource2() { /* ... */ return "B"; }
    static String fetchFromSource3() { /* ... */ return "C"; }
}

Co się dzieje:

• Gdy tylko jedno zadanie zakończy się pomyślnie — pozostałe są anulowane.
• Jeśli wszystkie się nie powiodą — zostaje rzucony wyjątek.

4. Automatyczne anulowanie i degradacja

StructuredTaskScope sam dba o anulowanie pozostałych zadań, jeśli wymaga tego polityka. Na przykład, jeśli jedno zadanie się nie powiodło (ShutdownOnFailure) lub jedno zakończyło się pomyślnie (ShutdownOnSuccess), pozostałe zadania dostają sygnał anulowania (interrupt).

Przykład: poprawne zakończenie z limitem czasu

import jdk.incubator.concurrent.StructuredTaskScope;
import java.time.Instant;
import java.util.concurrent.Future;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> f1 = scope.fork(() -> fetchWithTimeout());
    Future<String> f2 = scope.fork(() -> fetchWithTimeout());

    scope.joinUntil(Instant.now().plusSeconds(2)); // czekamy maksymalnie 2 sekundy
    scope.throwIfFailed();

    String result = f1.resultNow() + f2.resultNow();
    System.out.println(result);
}

Jeżeli zadania nie zakończyły się w ciągu 2 sekundy — zostanie rzucony wyjątek, a wszystkie zadania zostaną anulowane.

5. Błędy i obsługa wyjątków

Jak wyjątki podzadań są kierowane do scope

Czasem w trakcie koncertu ktoś jednak chybi nutę — StructuredTaskScope nie udaje, że nic się nie stało. Starannie zapisuje, kto zagrał nieczysto, a potem przekazuje dyrygentowi pełny raport. Gdy wywołujesz throwIfFailed(), rzuca agregowany wyjątek — coś w rodzaju zbiorczego raportu: „Oto lista tych, którzy dziś zafałszowali”. Jeśli trzeba, można rozwinąć to „drzewo przyczyn” i zobaczyć, kto dokładnie zawiódł. A jeśli chcesz dowiedzieć się o konkretnym wykonawcy — Future.exceptionNow() powie, jak skończyła się jego partia.

Kiedy anulowanie to nie porażka

Ważne: anulowanie zadania nie zawsze oznacza błąd. Jeśli dyrygent powiedział „koniec koncertu”, muzycy po prostu odkładają instrumenty — to cancelled, ale nie failed. Błędem jest tylko sytuacja, gdy ktoś faktycznie zagrał nie to, co trzeba — i ten wyjątek trafi już do zbiorczej informacji.

Przykład: drzewo przyczyn

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> f1 = scope.fork(() -> { throw new RuntimeException("Błąd 1"); });
    Future<String> f2 = scope.fork(() -> { throw new RuntimeException("Błąd 2"); });

    scope.join();
    scope.throwIfFailed(); // rzuci wyjątek z obiema przyczynami
} catch (Exception e) {
    e.printStackTrace();
    // Można pobrać wyjątki suppressed przez e.getSuppressed()
}

6. Porównanie z CompletableFuture

StructuredTaskScope i CompletableFuture — oba pozwalają uruchamiać zadania równolegle, ale:

• StructuredTaskScope jest wygodny, gdy zadania są logicznie powiązane i powinny kończyć się/anulować razem (hierarchia zadań).
• CompletableFuture sprawdza się przy kompozycji zadań bez hierarchii (np. łańcuchy przekształceń, scenariusze reaktywne).

Kiedy StructuredTaskScope upraszcza kod:

• Gdy trzeba zagwarantować, że wszystkie podzadania zakończą się przed wyjściem z bloku.
• Gdy potrzebne jest scentralizowane anulowanie i obsługa błędów.
• Gdy ważne jest, aby nie zostały żadne „wiszące” zadania.

Kiedy CompletableFuture jest wygodniejszy:

• Gdy zadania nie są powiązane i mogą żyć własnym życiem.
• Gdy potrzebna jest złożona kompozycja (thenCombine, thenCompose itd.).

7. Praktyka: agregator żądań HTTP

Zadanie: wysłać żądania do 3 źródeł, pobrać pierwszą pomyślną odpowiedź

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

public class HttpAggregator {
    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
            Future<String> f1 = scope.fork(() -> httpRequest("https://api1.example.com"));
            Future<String> f2 = scope.fork(() -> httpRequest("https://api2.example.com"));
            Future<String> f3 = scope.fork(() -> httpRequest("https://api3.example.com"));

            scope.join();
            scope.throwIfFailed();

            String result = scope.result();
            System.out.println("Pierwsza pomyślna odpowiedź: " + result);
        }
    }

    static String httpRequest(String url) throws Exception {
        // Symulacja żądania (można użyć HttpClient)
        Thread.sleep((long) (Math.random() * 1000));
        if (Math.random() < 0.3) throw new RuntimeException("Błąd żądania: " + url);
        return "Odpowiedź z " + url;
    }
}

Zadanie: jeśli jedno podzadanie padło — poprawnie wygasić pozostałe

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> f1 = scope.fork(() -> httpRequest("https://api1.example.com"));
    Future<String> f2 = scope.fork(() -> httpRequest("https://api2.example.com"));

    scope.join();
    scope.throwIfFailed();

    String result = f1.resultNow() + f2.resultNow();
    System.out.println("Obie odpowiedzi: " + result);
} catch (Exception e) {
    System.err.println("Błąd w jednym z zadań: " + e.getMessage());
}

8. Typowe błędy przy pracy z StructuredTaskScope

Błąd nr 1: zapomniano wywołać join() lub throwIfFailed().
Jeśli nie wywołasz join(), zadania mogą nie zakończyć się przed wyjściem z bloku. Jeśli nie wywołasz throwIfFailed(), błędy podzadań pozostaną niezauważone.

Błąd nr 2: próba pobrania wyniku przed zakończeniem zadania.
Wywołanie resultNow() przed zakończeniem zadania rzuci IllegalStateException. Najpierw poczekaj na zakończenie przez join().

Błąd nr 3: ignorowanie anulowania.
Jeśli zadanie zostało anulowane (np. z powodu polityki scope), nie próbuj pobierać jego wyniku — będzie wyjątek.

Błąd nr 4: mieszanie różnych polityk zakończenia.
Nie warto ręcznie anulować zadania wewnątrz scope — używaj polityk ShutdownOnFailure lub ShutdownOnSuccess.

Błąd nr 5: uruchamianie długich zadań CPU-bound w wirtualnych wątkach.
StructuredTaskScope domyślnie używa wirtualnych wątków — są idealne dla zadań I/O-bound, ale nie przyspieszą ciężkich obliczeń.

Błąd nr 6: zapomniano zamknąć scope (brak try-with-resources).
StructuredTaskScope implementuje AutoCloseable — zawsze używaj try-with-resources, aby zagwarantować zakończenie wszystkich zadań.