Concurrence structurée en Java

La concurrence structurée traite un groupe de sous-tâches concurrentes comme une seule unité de travail : elles sont lancées ensemble, elles se terminent ensemble, et si l'une échoue ou si l'appelant est annulé, les autres le sont aussi — aucun thread orphelin ne survit au bloc qui les a démarrés. Le modèle est fourni par java.util.concurrent.StructuredTaskScope (une API en prévisualisation introduite dans Java 21) et repose sur les mêmes threads virtuels abordés plus tôt dans cette partie. L'objectif est simple : rendre le code concurrent aussi facile à lire, déboguer et raisonner qu'une méthode séquentielle ordinaire.

Ce chapitre explique pourquoi le terme « structuré » est important, l'anatomie d'un scope de tâches, les deux politiques d'arrêt intégrées, la propagation des délais et des annulations, ainsi qu'un exemple pratique. Il suppose que vous êtes à l'aise avec le framework executor et Callable/Future.

Pourquoi « structuré » ?

Les pools de threads classiques sont non structurés : vous soumettez une tâche à un ExecutorService partagé et récupérez un Future dont la durée de vie est indépendante de la méthode qui l'a créé. Une tâche peut survivre à son appelant, une erreur dans une tâche est invisible pour ses tâches sœurs, et l'annulation doit être câblée manuellement. Le résultat : des threads fuités et une gestion des erreurs enchevêtrée.

La concurrence structurée emprunte la discipline du flux de contrôle structuré : tout comme un bloc try délimite ses instructions, un scope de tâches confine ses sous-tâches. Les sous-tâches lancées à l'intérieur d'un bloc doivent toutes se terminer avant que le bloc ne se ferme. Les durées de vie s'imbriquent proprement, ce qui fait qu'un vidage de threads et une trace de pile vous indiquent réellement qui a démarré quoi.

Problématique	Non structuré (pool partagé `ExecutorService`)	Structuré (`StructuredTaskScope`)
Durée de vie des sous-tâches	Indépendante de l'appelant	Délimitée par le bloc englobant
Erreur dans une sous-tâche	Cachée dans un `Future` jusqu'à l'appel de `get`	Peut court-circuiter tout le scope
Annulation	Manuelle, facile à oublier	Automatique en cas d'échec ou d'interruption
Nettoyage des ressources	À votre charge	`close()` attend toutes les sous-tâches

La forme d'un scope

Un scope est un AutoCloseable, il vit donc dans un bloc try-with-resources. Vous forkez des sous-tâches (chacune renvoie un handle Subtask), appelez join() pour les attendre, puis lisez chaque résultat. La politique ShutdownOnFailure annule les sous-tâches restantes dès que l'une d'elles lève une exception :

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

Si fetchUser lève une exception, ShutdownOnFailure interrompt le fetchOrderCount encore en cours, join() retourne, et throwIfFailed() relance la cause originale encapsulée dans une ExecutionException. Vous ne laissez jamais fuiter de thread.

Politiques d'arrêt intégrées

Les deux politiques fournies couvrent les cas courants ; vous sous-classez StructuredTaskScope pour tout autre cas.

Politique	Se termine quand	Utilisation
`ShutdownOnFailure`	Toutes réussissent, ou l'une échoue	Fan-out où vous avez besoin de chaque résultat (le cas courant)
`ShutdownOnSuccess<T>`	Premier succès, ou toutes échouent	Mise en compétition de sources redondantes ; prendre la réponse la plus rapide

ShutdownOnSuccess renvoie le gagnant via result() et annule les perdants :

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
}

Propagation des délais et des annulations

Un scope peut être joint avec un délai limite ; lorsqu'il s'écoule, les sous-tâches non terminées sont annulées :

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

L'annulation est coopérative et se propage vers le bas : si le thread propriétaire du scope est interrompu, chaque sous-tâche est interrompue à son tour. Étant donné que chaque sous-tâche s'exécute sur son propre thread virtuel, en lancer des milliers est bon marché — le scope, et non une taille de pool fixe, est l'unité sur laquelle vous raisonnez.

Exemple pratique : fan-out, échec et jonction d'une liste

StructuredTaskScope est une fonctionnalité en prévisualisation ; pour que cet exemple soit exécutable sur un JDK stable, nous modélisons la même idée avec un executor un-thread-virtuel-par-tâche : un bloc try-with-resources qui délimite un groupe de sous-tâches et ne se ferme que lorsque chaque thread de sous-tâche s'est terminé. Il lance deux appels en parallèle, puis montre comment un échec court-circuite l'unité de travail et comment invokeAll joint une liste entière en une seule opération.

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

Ce qu'il faut retenir de l'exécution :

Les deux sous-tâches ont signalé is virtual : true — chaque submit s'est exécuté sur son propre thread virtuel, le même porteur léger qu'utilise StructuredTaskScope.fork, donc créer un thread par sous-tâche est peu coûteux.
Le bloc du chemin heureux a affiché ran concurrently (<320ms): true alors que les deux fetch dorment 120ms et 200ms : ils se sont chevauchés, donc le temps réel correspond à la branche la plus lente (~200ms), et non à la somme (320ms). Ce chevauchement est tout l'intérêt du fan-out.
Quitter le bloc try-with-resources a appelé close(), qui a bloqué jusqu'à ce que chaque thread de sous-tâche se termine — le scope est l'unité de durée de vie, exactement la discipline qu'impose StructuredTaskScope par construction.
Dans la section d'échec, le programme a affiché caught: IllegalStateException -> upstream said no : une erreur levée à l'intérieur d'une sous-tâche remonte au point de jonction encapsulée dans ExecutionException, et getCause() vous restitue l'exception originale.
Après avoir capturé l'échec, il a affiché sibling cancelled: true — nous avons annulé la branche good encore en cours d'exécution afin qu'aucun orphelin ne survive au bloc, ce qui est précisément ce que ShutdownOnFailure fait automatiquement pour vous ; ici nous l'avons fait à la main pour montrer le mécanisme.

Sujets connexes

Threads virtuels — les threads légers sur lesquels s'exécute chaque sous-tâche.
Threads virtuels modernes — modèles pratiques et pièges à éviter.
Framework executor — la base non structurée que ce modèle remplace.
Callable et Future — les types de tâche et de résultat utilisés au point de jonction.
CompletableFuture — composer des résultats asynchrones sans blocage aux jonctions.

Pratique

Avec StructuredTaskScope.ShutdownOnFailure, que se passe-t-il avec les autres sous-tâches lancées lorsque l'une d'elles lève une exception ?

Les sous-tâches restantes sont annulées, et join() suivi de throwIfFailed() relance l'échec original encapsulé dans une ExecutionExceptionLes autres sous-tâches continuent de s'exécuter indépendamment et vous devez annuler chacune vous-même après la fermeture du blocLa JVM entière se termine immédiatement car une exception non capturée s'est échappée d'un threadRien ne se passe tant que vous n'interrogez pas chaque handle Subtask, les échecs étant stockés silencieusement comme dans un Future ordinaire