Concorrenza strutturata in Java

La concorrenza strutturata tratta un gruppo di sottocompiti concorrenti come un'unica unità di lavoro: vengono avviati insieme, terminano insieme e, se uno fallisce o il chiamante viene annullato, gli altri vengono annullati a loro volta — nessun thread orfano sopravvive al blocco che lo ha creato. Il modello è fornito da java.util.concurrent.StructuredTaskScope (una API di anteprima introdotta in Java 21) e si basa sugli stessi thread virtuali trattati in precedenza in questa sezione. L'obiettivo è semplice: rendere il codice concorrente facile da leggere, debuggare e comprendere quanto un normale metodo sequenziale.

Questo capitolo spiega perché il termine "strutturato" è importante, l'anatomia di un task scope, le due policy di shutdown integrate, come le scadenze e la cancellazione si propagano, e un esempio pratico eseguibile. Si presuppone che tu abbia familiarità con il framework executor e con Callable/Future.

Perché "strutturato"?

I thread pool classici sono non strutturati: si invia un task a un ExecutorService condiviso tramite submit e si ottiene un Future la cui durata di vita è indipendente dal metodo che lo ha creato. Un task può sopravvivere al suo chiamante, un errore in un task è invisibile ai suoi fratelli e la cancellazione deve essere gestita manualmente. Il risultato sono thread in perdita e una gestione degli errori aggrovigliata.

La concorrenza strutturata prende in prestito la disciplina del flusso di controllo strutturato: proprio come un blocco try delimita le sue istruzioni, un task scope confina i suoi sottocompiti. I sottocompiti avviati all'interno di un blocco devono tutti completarsi prima che il blocco esca. Le durate di vita si annidano in modo ordinato, così un thread dump e uno stack trace indicano chiaramente chi ha avviato cosa.

Aspetto	Non strutturato (pool condiviso `ExecutorService`)	Strutturato (`StructuredTaskScope`)
Durata dei sottocompiti	Indipendente dal chiamante	Limitata dal blocco che li contiene
Errore in un sottocompito	Nascosto in un `Future` finché non si chiama `get`	Può interrompere l'intero scope
Cancellazione	Manuale, facile da dimenticare	Automatica in caso di fallimento o interruzione
Pulizia delle risorse	A tuo carico	`close()` attende ogni sottocompito

La struttura di uno scope

Uno scope è un AutoCloseable, quindi vive in un blocco try-with-resources. Si usano fork per avviare i sottocompiti (ciascuno restituisce un handle Subtask), si chiama join() per attendere il completamento, poi si legge ogni risultato. La policy ShutdownOnFailure annulla i sottocompiti rimanenti nel momento in cui uno qualsiasi di essi lancia un'eccezione:

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

Se fetchUser lancia un'eccezione, ShutdownOnFailure interrompe il fetchOrderCount ancora in esecuzione, join() ritorna e throwIfFailed() rilancia la causa originale avvolta in un ExecutionException. Nessun thread va in perdita.

Policy di shutdown integrate

Le due policy fornite coprono i casi d'uso comuni; per tutto il resto si può creare una sottoclasse di StructuredTaskScope.

Policy	Termina quando	Usala per
`ShutdownOnFailure`	Tutti hanno successo, o uno fallisce	Fan-out in cui servono tutti i risultati (il caso più comune)
`ShutdownOnSuccess<T>`	Primo successo, o tutti falliscono	Gara tra sorgenti ridondanti; prendi la risposta più veloce

ShutdownOnSuccess restituisce il vincitore tramite result() e annulla i perdenti:

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
}

Le scadenze e la cancellazione si propagano

È possibile attendere uno scope con una scadenza; quando essa scade, i sottocompiti non terminati vengono annullati:

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

La cancellazione è cooperativa e si propaga verso il basso: se il thread che possiede lo scope viene interrotto, ogni sottocompito viene interrotto a sua volta. Poiché ogni sottocompito viene eseguito sul proprio thread virtuale, avviarne migliaia è economico — lo scope, non la dimensione di un pool fisso, è l'unità su cui ragionare.

Un esempio pratico: fan-out, fallimento e join di una lista

StructuredTaskScope è una funzionalità in anteprima, quindi per mantenere questo esempio eseguibile su un JDK stabile modelliamo la stessa idea con un executor virtual-thread-per-task: un blocco try-with-resources che delimita un gruppo di sottocompiti e termina solo quando ogni thread sottocompito è terminato. Esegue due chiamate in modo concorrente, poi mostra come un fallimento interrompe l'unità di lavoro e come invokeAll attende un'intera lista contemporaneamente.

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");
  }
}

Cosa osservare dall'esecuzione:

Entrambi i sottocompiti hanno riportato is virtual : true — ogni submit è stato eseguito sul proprio thread virtuale, lo stesso carrier leggero usato da StructuredTaskScope.fork, quindi avviare un thread per sottocompito è economico.
Il blocco del percorso felice ha stampato ran concurrently (<320ms): true anche se le due fetch dormono 120ms e 200ms: si sono sovrapposte, quindi il tempo reale segue il ramo più lento (~200ms) e non la somma (320ms). Questa sovrapposizione è l'intero scopo del fan-out.
L'uscita dal blocco try-with-resources ha chiamato close(), che si è bloccato fino a quando ogni thread sottocompito non è terminato — lo scope è l'unità di durata di vita, esattamente la disciplina che StructuredTaskScope applica per costruzione.
Nella sezione del fallimento, il programma ha stampato caught: IllegalStateException -> upstream said no: un errore lanciato all'interno di un sottocompito emerge al punto di join avvolto in ExecutionException, e getCause() restituisce l'eccezione originale.
Dopo aver catturato il fallimento ha stampato sibling cancelled: true — abbiamo annullato il ramo good ancora in esecuzione in modo che nessun orfano sopravvivesse al blocco, che è esattamente ciò che ShutdownOnFailure fa automaticamente; qui lo abbiamo fatto manualmente per mostrare il meccanismo.

Argomenti correlati

Thread virtuali — i thread leggeri su cui viene eseguito ogni sottocompito.
Thread virtuali moderni — pattern pratici e insidie.
Framework executor — la base non strutturata che questo modello sostituisce.
Callable e Future — i tipi di task e risultato usati al punto di join.
CompletableFuture — composizione di risultati asincroni senza join bloccanti.

Pratica

Con StructuredTaskScope.ShutdownOnFailure, cosa succede agli altri sottocompiti avviati quando uno di essi lancia un'eccezione?

I sottocompiti rimanenti vengono annullati, e join() seguito da throwIfFailed() rilancia il fallimento originale avvolto in un ExecutionExceptionGli altri sottocompiti continuano a girare indipendentemente e devi annullarli tu stesso uno per uno dopo l'uscita dal bloccoL'intera JVM esce immediatamente perché un'eccezione non catturata è uscita da un threadNon succede nulla finché non esegui il polling di ogni handle Subtask, poiché i fallimenti vengono memorizzati in silenzio come in un normale Future