9. La programmation asynchrone

Le chapitre précédent a présenté les threads : on peut faire exécuter plusieurs morceaux de code « en même temps », les synchroniser, et confier des tâches à un pool de threads par l'intermédiaire d'un exécuteur ([ExecutorService]). Ce chapitre traite d'une question voisine mais différente : comment écrire un programme qui attend beaucoup (le disque, le réseau, une base de données) sans immobiliser de threads pendant ces attentes.

Java n'a pas de mots clés du langage pour cela (d'autres langages, comme C# ou JavaScript, ont les mots clés [async] et [await]). Il apporte deux réponses, qui se complètent et que ce chapitre présente toutes les deux :

• la classe [CompletableFuture<T>] (Java 8, enrichie depuis) représente une opération en cours dont on compose la suite par des méthodes de rappel : « quand le résultat sera là, fais ceci, puis cela » ([thenApply], [thenCompose], [thenCombine], [allOf]…). Aucun thread n'attend : c'est la fin de l'opération qui déclenche la suite ;
• les threads virtuels (« virtual threads », Java 21) permettent au contraire d'écrire du code bloquant, séquentiel, qui se lit de haut en bas : lorsqu'un thread virtuel attend une entrée / sortie, la JVM le décroche du thread du système qui l'exécutait (son thread porteur, « carrier thread ») et libère ce dernier pour un autre thread virtuel. On peut ainsi avoir des centaines de milliers de threads virtuels pour quelques threads du système. C'est la réponse moderne de Java au problème que d'autres langages résolvent avec [async] / [await].

Ces outils sont aujourd'hui partout : les chapitres suivants les retrouveront pour accéder aux bases de données, écrire des serveurs TCP qui servent chaque client par un thread virtuel, des clients web avec [java.net.http.HttpClient] et des services web avec Spring Boot.

Tous les exemples du chapitre sont des fichiers source compacts ([Async01.java], [Async02.java]…) que l'on exécute avec [java AsyncNN.java] dans le dossier [09-async]. Beaucoup d'entre eux mesurent des durées avec [System.nanoTime()] : les millisecondes affichées varient d'une exécution à l'autre et d'une machine à l'autre (les exemples ont été exécutés sur une machine à 2 processeurs). Ce sont les ordres de grandeur qui comptent : 1 seconde ou 10 secondes, 300 ms ou 0 ms.

9.1. Pourquoi l'asynchronisme

9.1.1. Les opérations qui attendent

Un programme fait deux sortes de travail :

• du calcul, qui occupe le processeur : trier une liste, calculer un impôt, analyser un texte ;
• des entrées / sorties (E/S), qui consistent pour l'essentiel à attendre un périphérique : lire un fichier sur le disque, envoyer une requête à un serveur web et attendre sa réponse, exécuter une requête SQL sur un serveur de bases de données.

Les ordres de grandeur sont très différents. Un processeur exécute quelques milliards d'instructions par seconde. Une lecture sur un disque SSD prend de l'ordre d'un dixième de milliseconde, un aller-retour sur un réseau local une fraction de milliseconde, un aller-retour vers un serveur situé à l'autre bout du pays quelques dizaines de millisecondes, une requête SQL un peu complexe parfois plusieurs secondes. Pendant ces attentes, le processeur aurait pu exécuter des millions d'instructions.

Lorsqu'un programme lit une ligne sur une connexion réseau avec une méthode bloquante, le thread qui exécute cette lecture est bloqué : il ne fait rien, mais il existe. Un thread Java « classique », que l'on appelle aujourd'hui thread de plate-forme (« platform thread »), est un thread du système d'exploitation : il occupe de la mémoire (sa pile, de l'ordre du mégaoctet), il est connu du système, et surtout il n'est plus disponible pour faire autre chose. Ce n'est pas gênant dans un petit programme console qui fait une chose à la fois. Cela l'est beaucoup plus dans deux situations :

• dans un serveur : chaque client servi par un thread bloqué immobilise ce thread. Si le pool de threads du serveur a N threads et que les N threads font une opération bloquante, il n'y a plus de thread disponible pour un nouveau client. Un serveur moderne sert des milliers de requêtes simultanées : c'est possible parce que ses attentes n'immobilisent pas de thread du système ;
• dans une interface graphique : un seul thread, le thread de l'interface (l'« Event Dispatch Thread » de Swing, le « JavaFX Application Thread »), traite les clics et redessine les fenêtres. S'il est bloqué par un téléchargement, la fenêtre se fige.

L'asynchronisme consiste à lancer l'opération d'E/S, à libérer le thread du système pendant qu'elle se déroule (le matériel et le système d'exploitation n'ont pas besoin de lui pour transférer les données), puis à reprendre le travail lorsque l'opération est terminée, éventuellement sur un autre thread du système :

• en [1], le thread de plate-forme T1 lance l'E/S et reste bloqué jusqu'à sa fin ;
• en [2], le thread virtuel V1 lance l'E/S ; la JVM le décroche de son thread porteur P1, qui peut aussitôt exécuter d'autres threads virtuels. Lorsque l'E/S se termine, la JVM raccroche V1 à un thread porteur libre, P2 (qui peut d'ailleurs être P1), et V1 continue son exécution juste après l'instruction bloquante, comme si de rien n'était.

Le programme [Async01.java] mesure la différence sur 100 opérations qui attendent chacune une seconde (l'attente simule une E/S). Il utilise des instructions que nous expliquerons dans la suite du chapitre ([Executors.newVirtualThreadPerTaskExecutor], [CompletableFuture]) : on s'intéresse ici surtout aux résultats.

// pourquoi l'asynchronisme : 100 opérations qui attendent chacune une seconde
// (une attente simule une E/S : lecture disque, requête réseau, requête SQL…)
import java.lang.management.ManagementFactory;

final int nbOperations = 100;
long debut;

void main() throws InterruptedException {
    IO.println(nbOperations + " opérations d'une seconde, "
            + Runtime.getRuntime().availableProcessors() + " processeurs");

    // 1 - un thread de plate-forme par opération, chaque thread est bloqué pendant l'attente
    debut = System.nanoTime();
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < nbOperations; i++) {
        threads.add(Thread.ofPlatform().start(() -> attendre(1000)));
    }
    afficher("1 - un thread de plate-forme par opération", threadsDeLaJvm());
    for (Thread thread : threads) {
        thread.join();
    }
    afficher("    fin", -1);

    // 2 - les opérations sont confiées à un pool de 10 threads, mais elles le bloquent
    debut = System.nanoTime();
    try (ExecutorService pool = Executors.newFixedThreadPool(10)) {
        for (int i = 0; i < nbOperations; i++) {
            pool.submit(() -> attendre(1000));
        }
        afficher("2 - pool de 10 threads, attentes bloquantes", threadsDeLaJvm());
    } // close() attend la fin de toutes les opérations
    afficher("    fin", -1);

    // 3 - un thread virtuel par opération : il libère son thread porteur pendant l'attente
    debut = System.nanoTime();
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < nbOperations; i++) {
            executeur.submit(() -> attendre(1000));
        }
        afficher("3 - un thread virtuel par opération", threadsDeLaJvm());
    }
    afficher("    fin", -1);

    // 4 - attentes asynchrones : aucun thread n'attend, une minuterie termine chaque opération
    debut = System.nanoTime();
    var operations = new CompletableFuture<?>[nbOperations];
    for (int i = 0; i < nbOperations; i++) {
        operations[i] = new CompletableFuture<Void>().completeOnTimeout(null, 1000, TimeUnit.MILLISECONDS);
    }
    afficher("4 - attentes asynchrones (CompletableFuture)", threadsDeLaJvm());
    CompletableFuture.allOf(operations).join();
    afficher("    fin", -1);
}

// une attente bloquante de [ms] millisecondes
void attendre(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

// nombre de threads de plate-forme de la JVM, une demi-seconde après le lancement des opérations
int threadsDeLaJvm() {
    long ecoule = (System.nanoTime() - debut) / 1_000_000;
    if (ecoule < 500) {
        attendre(500 - ecoule);
    }
    return ManagementFactory.getThreadMXBean().getThreadCount();
}

// affichage du temps écoulé et, s'il est fourni (>= 0), du nombre de threads
void afficher(String message, int nbThreads) {
    String infoThreads = nbThreads < 0 ? "" : String.format(", %3d threads dans la JVM", nbThreads);
    long ecoule = (System.nanoTime() - debut) / 1_000_000;
    IO.println(String.format("%-44s : %5d ms%s", message, ecoule, infoThreads));
}

Résultats :

> java Async01.java
100 opérations d'une seconde, 2 processeurs
1 - un thread de plate-forme par opération   :   534 ms, 106 threads dans la JVM
    fin                                      :  1061 ms
2 - pool de 10 threads, attentes bloquantes  :   501 ms,  16 threads dans la JVM
    fin                                      : 10021 ms
3 - un thread virtuel par opération          :   501 ms,  10 threads dans la JVM
    fin                                      :  1091 ms
4 - attentes asynchrones (CompletableFuture) :   500 ms,  11 threads dans la JVM
    fin                                      :  1003 ms

Commentaires :

• lignes 12-22 : la solution la plus simple, un thread de plate-forme par opération. [Thread.ofPlatform().start(tâche)] (Java 21) crée et démarre un thread, comme [new Thread(tâche).start()]. Chaque thread exécute [Thread.sleep(1000)] : il est bloqué pendant une seconde. Les 100 attentes se déroulent en même temps et l'ensemble dure bien une seconde (ligne 4 des résultats), mais au prix de 100 threads du système supplémentaires : la JVM en compte 106 (ligne 3). Avec 10 000 opérations, il faudrait 10 000 threads du système ;
• lignes 24-32 : les opérations sont confiées à un pool de 10 threads ([Executors.newFixedThreadPool(10)]), une taille raisonnable pour des calculs. Mais ici ses threads ne calculent pas : ils dorment. Les opérations attendent un thread libre, par vagues de 10, et l'ensemble dure… dix secondes (ligne 6). C'est ce qu'on appelle la famine du pool de threads (« thread pool starvation »), le défaut le plus fréquent des serveurs qui mélangent pool de threads et E/S bloquantes. La méthode [close()] d'un exécuteur (Java 19), appelée à la sortie du [try], refuse les nouvelles tâches et attend la fin de celles qui ont été soumises ;
• lignes 34-42 : un thread virtuel par opération. [Executors.newVirtualThreadPerTaskExecutor()] (Java 21) rend un exécuteur qui crée un nouveau thread virtuel pour chaque tâche soumise. Le code des tâches est le même qu'à l'étape 1 : un [Thread.sleep(1000)] bloquant. Mais un thread virtuel bloqué n'immobilise pas de thread du système : la JVM le décroche de son thread porteur. Les 100 attentes durent ensemble une seconde (ligne 8), et la JVM ne compte que 10 threads du système (ligne 7) : ses threads de service, et les deux threads porteurs, un par processeur, qui ont exécuté les 100 threads virtuels. Les threads virtuels ne figurent pas dans ce décompte : ce sont des objets de la JVM, pas des threads du système ;
• lignes 44-52 : la même attente sans aucun thread qui attend. [new CompletableFuture<Void>()] crée une opération « en cours », et [completeOnTimeout(valeur, délai, unité)] (Java 9) demande qu'elle soit terminée avec la valeur indiquée au bout du délai. Une seule minuterie de la JVM gère les 100 délais. [CompletableFuture.allOf(...)] rend une opération qui se termine quand toutes les autres le sont, et [join()] attend sa fin. L'ensemble dure une seconde (ligne 10), avec un seul thread de plus qu'à l'étape 3, celui de la minuterie (ligne 9) ;
• lignes 55-62 : [Thread.sleep] peut lancer l'exception contrôlée [InterruptedException] si le thread est interrompu pendant son sommeil (nous en reparlerons au paragraphe sur l'annulation) ; la méthode [attendre] l'intercepte pour que les lambdas des lignes 16, 28 et 38 n'aient pas à la traiter ;
• lignes 64-71 : [ManagementFactory.getThreadMXBean().getThreadCount()] donne le nombre de threads de plate-forme vivants de la JVM. La classe [ManagementFactory] appartient au module [java.management] : un fichier source compact n'importe automatiquement que le module [java.base], d'où l'importation de la ligne 3.

Retenons la leçon : pour attendre, il ne faut pas bloquer de thread du système. L'asynchronisme ne rend pas une E/S plus rapide ; il permet d'en mener beaucoup de front avec peu de threads du système, et de garder disponibles les threads dont l'application a besoin. Java offre pour cela deux chemins : composer des [CompletableFuture] (étape 4), ou écrire du code bloquant ordinaire exécuté par des threads virtuels (étape 3).

9.1.2. Plusieurs modèles successifs

Java a proposé plusieurs façons d'écrire du code qui attend sans bloquer. On rencontre encore les plus anciennes dans du code existant, il est donc utile de les reconnaître.

Modèle	Apparition	Principe
[Future] et exécuteurs	Java 5 (2004)	[ExecutorService.submit] confie une tâche à un pool de threads et rend un [Future<T>], qui représente le résultat à venir. Mais la seule façon d'obtenir ce résultat est d'attendre, en bloquant, avec [get()].
méthodes de rappel de NIO.2	Java 7 (2011)	les canaux asynchrones ([AsynchronousSocketChannel], [AsynchronousFileChannel]) lancent une opération et reçoivent un objet de rappel ([CompletionHandler]) dont la méthode [completed] ou [failed] est appelée par un thread d'un pool à la fin de l'opération.
[CompletableFuture]	Java 8 (2014)	un [Future] que l'on peut terminer soi-même et dont on compose la suite par des méthodes de rappel ([thenApply], [thenCompose], [exceptionally]…) ; les flux réactifs [java.util.concurrent.Flow] (Java 9) en sont le prolongement pour des suites de valeurs. Les bibliothèques réactives (Project Reactor, utilisé par Spring WebFlux, RxJava) reposent sur les mêmes idées.
threads virtuels	Java 21 (2023)	du code bloquant ordinaire, exécuté par des threads virtuels : la JVM libère le thread du système pendant les attentes. C'est le modèle recommandé aujourd'hui pour les nouvelles applications.
concurrence structurée	aperçu (« preview ») de Java 21 à Java 25	[StructuredTaskScope] organise un groupe de sous-tâches exécutées par des threads virtuels comme un bloc de code, avec une gestion commune des erreurs et de l'annulation (voir la fin du chapitre).

Les méthodes de rappel ont un défaut : le code est découpé en morceaux. Ce qui s'écrit naturellement « se connecter, puis lire, puis afficher » devient une méthode qui lance la connexion, une méthode de rappel qui lance la lecture, une autre méthode de rappel qui affiche ; les variables doivent être partagées entre ces méthodes, les exceptions sont récupérées à des endroits différents, et une simple boucle (« lire tant qu'il y a des données ») devient une méthode de rappel qui se relance elle-même. [CompletableFuture] rend ce découpage plus lisible en enchaînant les étapes ; les threads virtuels le font disparaître : le code asynchrone a la forme du code synchrone, parce qu'il est du code synchrone.

Le programme [Async02.java] écrit le même client TCP selon les deux modèles extrêmes : avec les méthodes de rappel de NIO.2, puis avec du code bloquant exécuté par un thread virtuel. Il contient aussi un petit serveur, exécuté par un thread, qui répond à deux clients (la programmation réseau sera étudiée en détail au chapitre 12 ; il suffit ici de savoir qu'un client se connecte au serveur et lit le message que celui-ci lui envoie).

// le même client TCP écrit avec des méthodes de rappel (NIO.2, Java 7), puis avec un thread virtuel (Java 21)
final int port = 27500;
final InetAddress local = InetAddress.getLoopbackAddress();
final ByteBuffer tampon = ByteBuffer.allocate(1024);
final CountDownLatch fin = new CountDownLatch(1);

void main() throws IOException, InterruptedException {
    // un serveur minimal, exécuté par un thread : il sert deux clients puis s'arrête
    var ecoute = new ServerSocket(port, 10, local);
    Thread serveur = Thread.ofPlatform().name("serveur").start(() -> {
        try (ecoute) {
            for (int i = 1; i <= 2; i++) {
                try (Socket service = ecoute.accept()) {
                    // le serveur prend son temps pour répondre
                    Thread.sleep(500);
                    byte[] message = ("Bonjour client n° " + i).getBytes(StandardCharsets.UTF_8);
                    service.getOutputStream().write(message);
                }
            }
        } catch (IOException | InterruptedException e) {
            afficher("serveur : " + e);
        }
    });
    afficher("début du programme");

    // 1 - le style « méthodes de rappel » : AsynchronousSocketChannel et CompletionHandler
    var client = AsynchronousSocketChannel.open();
    // on lance la connexion en indiquant l'objet à rappeler quand elle sera faite
    client.connect(new InetSocketAddress(local, port), client, new Connecte());
    afficher("rappels : connexion lancée, le programme continue");
    // le thread principal n'a plus rien à faire : il attend le signal de la dernière méthode de rappel
    fin.await();

    // 2 - le style bloquant, exécuté par un thread virtuel : le code se lit de haut en bas
    Thread virtuel = Thread.ofVirtual().start(() -> {
        try (var client2 = new Socket(local, port)) {
            afficher("thread virtuel : connecté, lecture lancée");
            byte[] octets = new byte[1024];
            int nbOctets = client2.getInputStream().read(octets);
            String message = new String(octets, 0, nbOctets, StandardCharsets.UTF_8);
            afficher("thread virtuel : reçu [" + message + "]");
        } catch (IOException e) {
            afficher("thread virtuel : " + e);
        }
    });
    virtuel.join();
    serveur.join();
}

// méthode de rappel n° 1 : la connexion est faite (ou a échoué)
class Connecte implements CompletionHandler<Void, AsynchronousSocketChannel> {
    @Override
    public void completed(Void rien, AsynchronousSocketChannel canal) {
        afficher("rappels : connecté, lecture lancée");
        // on lance la lecture de la réponse du serveur avec un autre objet de rappel
        canal.read(tampon, canal, new Recu());
    }

    @Override
    public void failed(Throwable e, AsynchronousSocketChannel canal) {
        afficher("rappels : échec de la connexion : " + e);
        fin.countDown();
    }
}

// méthode de rappel n° 2 : des données sont arrivées
class Recu implements CompletionHandler<Integer, AsynchronousSocketChannel> {
    @Override
    public void completed(Integer nbOctets, AsynchronousSocketChannel canal) {
        afficher("rappels : reçu [" + new String(tampon.array(), 0, nbOctets, StandardCharsets.UTF_8) + "]");
        try {
            canal.close();
        } catch (IOException e) {
            afficher("rappels : " + e);
        }
        // on prévient le thread principal
        fin.countDown();
    }

    @Override
    public void failed(Throwable e, AsynchronousSocketChannel canal) {
        afficher("rappels : échec de la lecture : " + e);
        fin.countDown();
    }
}

// affichage préfixé par le thread qui l'exécute
// (nom d'un thread de plate-forme, description d'un thread virtuel)
void afficher(String message) {
    Thread thread = Thread.currentThread();
    IO.println("[" + (thread.isVirtual() ? thread.toString() : thread.getName()) + "] " + message);
}

Résultats :

> java Async02.java
[main] début du programme
[main] rappels : connexion lancée, le programme continue
[InnocuousThread-4] rappels : connecté, lecture lancée
[InnocuousThread-2] rappels : reçu [Bonjour client n° 1]
[VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1] thread virtuel : connecté, lecture lancée
[VirtualThread[#28]/runnable@ForkJoinPool-1-worker-2] thread virtuel : reçu [Bonjour client n° 2]

Commentaires :

• lignes 8-23 : le serveur. Il écoute sur le port 27500 de la machine locale ([InetAddress.getLoopbackAddress()], c'est-à-dire 127.0.0.1), accepte deux connexions l'une après l'autre et envoie à chaque client, après une demi-seconde, un message de bienvenue. Il est exécuté par un thread de plate-forme nommé [serveur]. La forme [try (ecoute)] (Java 9) fermera à la fin du bloc le [ServerSocket] déclaré à la ligne 9 ;
• lignes 26-32 : le client écrit avec des méthodes de rappel :
  • ligne 27 : un [AsynchronousSocketChannel] est un canal TCP dont les opérations sont asynchrones ;
  • ligne 29 : [connect] lance la connexion et rend la main immédiatement (ligne 3 des résultats). Ses deux derniers paramètres sont un objet quelconque que la méthode de rappel recevra (ici le canal lui-même) et l'objet de rappel, de type [Connecte] ;
  • ligne 32 : le thread principal n'a plus rien à faire. Pour que le programme ne se termine pas avant la fin de l'échange, il attend un signal que donnera la dernière méthode de rappel : un [CountDownLatch] initialisé à 1 (ligne 5), dont la méthode [await] bloque le thread qui l'appelle jusqu'à ce qu'un autre thread appelle sa méthode [countDown] ;
• lignes 50-64 : l'objet de rappel de la connexion implémente l'interface [CompletionHandler<V, A>], où [V] est le type du résultat de l'opération ([Void] pour une connexion) et [A] celui de l'objet transmis. Sa méthode [completed] est appelée si l'opération réussit, sa méthode [failed] si elle échoue. Une fois connecté, on lance la lecture avec [read] en indiquant un second objet de rappel, de type [Recu] (ligne 56) ;
• lignes 66-85 : l'objet de rappel de la lecture reçoit le nombre d'octets lus, affiche le message, ferme le canal et débloque le thread principal. Les classes [Connecte] et [Recu] sont des classes internes de la classe implicite du fichier compact : elles ont accès à ses champs [tampon] et [fin] et à sa méthode [afficher] ;
• les résultats montrent que les méthodes de rappel sont exécutées par d'autres threads que le thread principal (des threads du pool de la JVM consacré aux canaux asynchrones, nommés [InnocuousThread-N]) ;
• lignes 34-46 : le même client, écrit avec un [Socket] ordinaire et des opérations bloquantes ([new Socket(...)] se connecte, [read] attend les données), exécuté par un thread virtuel créé par [Thread.ofVirtual().start(tâche)] (Java 21). Le code se lit de haut en bas comme du code synchrone, les variables sont des variables locales ordinaires, les erreurs sont traitées par un seul [try / catch], il n'y a ni méthode de rappel ni signal à gérer ;
• lignes 87-92 : chaque affichage est préfixé par le thread qui l'exécute. Pour un thread virtuel, [toString()] rend une description de la forme [VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1] : le numéro du thread virtuel, son état et son thread porteur actuel. Les lignes 6 et 7 des résultats montrent le même thread virtuel n° 28, porté d'abord par [worker-1] puis par [worker-2] : pendant l'attente de la réponse du serveur, il a été décroché de son premier porteur et raccroché ensuite à un autre.

Il n'est pas nécessaire de retenir l'interface [CompletionHandler]. Si l'on doit utiliser une API qui ne propose que ce modèle, on la transforme en [CompletableFuture] (nous le ferons dans [Async17.java]), ou l'on utilise sa version bloquante sur un thread virtuel.

9.2. Les opérations à venir : Future et CompletableFuture

Un [Future<T>] (interface du paquetage [java.util.concurrent], Java 5) est un objet qui représente le résultat, de type [T], d'une opération en cours ou terminée. Le chapitre précédent l'a rencontré : c'est ce que rend la méthode [submit] d'un exécuteur. On peut l'interroger (est-elle terminée ? a-t-elle réussi ?), l'annuler ([cancel]) et attendre son résultat ([get]), mais rien d'autre : la seule façon d'exploiter le résultat est de bloquer un thread jusqu'à ce qu'il soit là.

La classe [CompletableFuture<T>] (Java 8) implémente [Future<T>] et ajoute deux possibilités essentielles :

• on peut la terminer soi-même, avec un résultat ([complete]) ou une exception ([completeExceptionally]) ;
• on peut lui attacher la suite du traitement, une fonction qui sera exécutée lorsque le résultat sera disponible ([thenApply], [thenAccept], [thenCompose]…), ce qui produit un nouveau [CompletableFuture]. On construit ainsi des chaînes d'opérations sans qu'aucun thread n'attende. Ces méthodes sont déclarées par l'interface [CompletionStage<T>], que [CompletableFuture] implémente également.

Un [CompletableFuture] n'est pas un thread. Il peut représenter un traitement exécuté par un thread d'un exécuteur (c'est le cas de ceux que crée [supplyAsync]), mais beaucoup ne sont exécutés par aucun thread : celui de l'étape 4 de [Async01.java] est simplement « terminé » par une minuterie au bout d'une seconde, celui que rend [HttpClient.sendAsync] est terminé lorsque la réponse du serveur est arrivée. C'est une promesse de résultat.

Le programme [Async03.java] présente les principales façons de créer, d'interroger et d'attendre un [CompletableFuture] :

// les CompletableFuture : création, attente, état, résultat, continuation
void main() throws InterruptedException {
    // 1 - supplyAsync confie un traitement à un exécuteur et rend un CompletableFuture<T>
    CompletableFuture<Long> f1 = CompletableFuture.supplyAsync(() -> {
        // un calcul de quelques dixièmes de seconde
        long somme = 0;
        for (int i = 1; i <= 200_000_000; i++) {
            somme += i % 7;
        }
        return somme;
    });
    IO.println("f1 lancé, état : " + f1.state());
    // join() attend la fin du traitement, en bloquant le thread courant, et rend son résultat
    long resultat = f1.join();
    IO.println("f1 terminé, état : " + f1.state() + ", résultat : " + resultat);

    // 2 - un CompletableFuture terminé par une minuterie au bout de 300 ms, sans occuper de thread
    CompletableFuture<Void> f2 = new CompletableFuture<Void>()
            .completeOnTimeout(null, 300, TimeUnit.MILLISECONDS);
    IO.println("f2 lancé, état : " + f2.state());
    f2.join();
    IO.println("f2 terminé, état : " + f2.state());

    // 3 - des CompletableFuture déjà terminés, avec succès ou en échec
    CompletableFuture<Integer> f3 = CompletableFuture.completedFuture(42);
    CompletableFuture<Integer> f4 = CompletableFuture.failedFuture(new IllegalStateException("panne"));
    IO.println("f3 : état " + f3.state() + ", résultat " + f3.resultNow()
            + " ; f4 : état " + f4.state() + ", exception " + f4.exceptionNow());

    // 4 - un traitement qui échoue : join() lance une CompletionException qui enveloppe son exception
    CompletableFuture<Integer> f5 = CompletableFuture.supplyAsync(() -> Integer.parseInt("abc"));
    try {
        f5.join();
    } catch (CompletionException e) {
        IO.println("f5 : état " + f5.state() + ", join() lance " + e.getClass().getSimpleName()
                + " contenant " + e.getCause());
    }

    // 5 - get(), hérité de l'interface Future, enveloppe l'exception dans une ExecutionException
    try {
        f5.get();
    } catch (ExecutionException e) {
        IO.println("f5 : get() lance " + e.getClass().getSimpleName()
                + " contenant " + e.getCause().getClass().getSimpleName());
    }

    // 6 - une continuation : ce qu'il faudra faire quand le CompletableFuture sera terminé
    CompletableFuture<Integer> f6 = f2.thenApply(rien -> 2 * 21);
    CompletableFuture<Void> f7 = f6.thenAccept(valeur -> IO.println("f7 : la continuation a reçu " + valeur));
    f7.join();

    // 7 - isDone, isCompletedExceptionally, isCancelled
    IO.println("f1 : isDone=" + f1.isDone() + ", f5 : isDone=" + f5.isDone()
            + ", isCompletedExceptionally=" + f5.isCompletedExceptionally()
            + ", isCancelled=" + f5.isCancelled());
}

Résultats :

> java Async03.java
f1 lancé, état : RUNNING
f1 terminé, état : SUCCESS, résultat : 599999998
f2 lancé, état : RUNNING
f2 terminé, état : SUCCESS
f3 : état SUCCESS, résultat 42 ; f4 : état FAILED, exception java.lang.IllegalStateException: panne
f5 : état FAILED, join() lance CompletionException contenant java.lang.NumberFormatException: For input string: "abc"
f5 : get() lance ExecutionException contenant NumberFormatException
f7 : la continuation a reçu 42
f1 : isDone=true, f5 : isDone=true, isCompletedExceptionally=true, isCancelled=false

Commentaires :

• lignes 4-11 : [CompletableFuture.supplyAsync(fournisseur)] confie une fonction sans paramètre ([Supplier<T>]) à un exécuteur et rend aussitôt un [CompletableFuture<T>] qui représente son exécution. La fonction rend un [long], on obtient donc un [CompletableFuture<Long>]. Sans exécuteur explicite, c'est le pool commun de la JVM, [ForkJoinPool.commonPool()], qui exécute la fonction ; nous verrons plus loin comment choisir l'exécuteur. Pour un traitement sans résultat ([Runnable]), on utilise [runAsync], qui rend un [CompletableFuture<Void>] ;
• ligne 12 : la méthode [state()] (Java 19) donne l'état de l'opération : [RUNNING] (en cours), [SUCCESS] (terminée avec succès), [FAILED] (terminée sur une exception) ou [CANCELLED] (annulée). Ce sont les constantes de l'énumération [Future.State] ; les trois dernières sont des états finaux : une opération terminée ne change plus d'état ;
• ligne 14 : [join()] attend la fin de l'opération et rend son résultat. Comme [Thread.join] au chapitre précédent, c'est une attente bloquante : le thread courant est immobilisé (le thread principal ici, ce qui ne gêne pas un petit programme console) ;
• lignes 17-22 : un [CompletableFuture] créé « en cours » par son constructeur et terminé par une minuterie au bout de 300 ms ([completeOnTimeout], Java 9). Aucun thread n'est occupé pendant ce temps : c'est l'équivalent asynchrone de [Thread.sleep(300)]. Son type est [CompletableFuture<Void>] : il ne porte pas de résultat ([Void] est la classe qui correspond à [void] ; sa seule valeur est [null]) ;
• lignes 25-28 : [completedFuture(42)] rend un [CompletableFuture] déjà terminé dont le résultat est 42, [failedFuture(exception)] (Java 9) un [CompletableFuture] déjà en échec. Ils servent lorsqu'une méthode doit rendre un [CompletableFuture] mais a déjà le résultat (une valeur en cache, par exemple) ou sait déjà qu'elle échoue. [resultNow()] et [exceptionNow()] (Java 19) rendent le résultat ou l'exception d'une opération terminée, sans attendre (elles lancent une [IllegalStateException] si l'opération n'est pas dans l'état voulu) ;
• lignes 31-37 : une opération dont la fonction lance une exception se termine dans l'état [FAILED]. L'exception est conservée dans le [CompletableFuture], et [join()] lance une [CompletionException] (une exception non contrôlée) dont la cause ([getCause()]) est l'exception d'origine, ici une [NumberFormatException] ;
• lignes 40-45 : [get()], la méthode héritée de l'interface [Future], attend elle aussi en bloquant, mais enveloppe l'exception dans une [ExecutionException], une exception contrôlée, et peut aussi lancer une [InterruptedException], contrôlée elle aussi : l'appel est donc placé dans un [try / catch] qui intercepte la première, et la méthode [main] déclare la seconde (ligne 2). [join()] est plus commode dans le code qui manipule des [CompletableFuture] ; [get()] est la seule disponible sur un simple [Future], et il en existe une version avec délai, [get(délai, unité)] ;
• lignes 48-50 : une continuation. [f2.thenApply(fonction)] rend un nouveau [CompletableFuture] qui sera terminé avec le résultat de la fonction appliquée au résultat de [f2], lorsque [f2] sera terminé. Ici la fonction ignore son paramètre ([null]) et rend 42 : on obtient un [CompletableFuture<Integer>]. [thenAccept] attache une fonction qui consomme le résultat sans en produire ([Consumer<T>]) et rend un [CompletableFuture<Void>]. Si l'opération d'origine échoue, la continuation n'est pas exécutée et l'échec se propage le long de la chaîne ;
• lignes 53-55 : les méthodes booléennes [isDone()] (terminée, quelle que soit la façon), [isCompletedExceptionally()] (terminée sur une exception ou annulée) et [isCancelled()] (annulée).

Les méthodes de [CompletableFuture] sont nombreuses (plus de cinquante). Le tableau suivant présente celles que nous utiliserons :

Méthode	Rôle
[supplyAsync(f)], [runAsync(r)]	exécuter une fonction (avec ou sans résultat) par un exécuteur
[completedFuture(v)], [failedFuture(e)]	créer une opération déjà terminée, avec succès ou en échec
[complete(v)], [completeExceptionally(e)]	terminer soi-même une opération
[thenApply(f)]	transformer le résultat : [T → U]
[thenAccept(c)], [thenRun(r)]	consommer le résultat, ou exécuter une action après l'opération
[thenCompose(f)]	enchaîner une autre opération asynchrone : [T → CompletableFuture<U>]
[thenCombine(autre, f)]	combiner les résultats de deux opérations menées de front
[allOf(...)], [anyOf(...)]	attendre toutes les opérations, ou la première terminée
[exceptionally(f)], [handle(f)], [whenComplete(c)]	traiter ou observer un échec
[orTimeout(...)], [completeOnTimeout(...)]	limiter la durée d'une opération
[join()], [get()]	attendre le résultat en bloquant le thread courant

Les méthodes [thenXxx], [handle] et [whenComplete] existent aussi en version [thenXxxAsync] : nous verrons la différence au paragraphe « Quel thread exécute la suite ? ».

9.3. Les méthodes asynchrones

9.3.1. Première façon : rendre un CompletableFuture

En Java, une méthode asynchrone est simplement une méthode qui lance une opération et rend aussitôt, sans attendre, un [CompletableFuture] qui la représente :

• [CompletableFuture<Void>] si l'opération ne rend pas de résultat ;
• [CompletableFuture<T>] si elle rend un résultat de type [T].

Le langage n'impose aucune convention de nommage. Les bibliothèques de Java suffixent souvent ces méthodes par [Async] ([sendAsync], [supplyAsync]), mais ce n'est pas une règle : le type de retour suffit à les reconnaître.

Le programme [Async04.java] prépare un café et compte des sucres :

// méthodes qui rendent un CompletableFuture : les étapes sont composées avec thenRun et thenCompose
final long debut = System.nanoTime();

void main() {
    afficher("appel de preparerCafe");
    CompletableFuture<Void> preparation = preparerCafe();
    afficher("preparerCafe a rendu la main : le programme continue");
    afficher("appel de compterSucres");
    CompletableFuture<Integer> comptage = compterSucres(3);
    afficher("compterSucres a rendu la main");
    // on combine les deux opérations, puis on attend le résultat de la combinaison
    int nbSucres = preparation.thenCombine(comptage, (rien, nombre) -> nombre).join();
    afficher("café prêt, avec " + nbSucres + " sucre(s)");
}

// une opération qui ne rend pas de résultat : elle rend un CompletableFuture<Void>
CompletableFuture<Void> preparerCafe() {
    afficher("  café : la machine chauffe");
    // la méthode décrit l'enchaînement des étapes, puis rend la main aussitôt
    return delai(1000)
            .thenRun(() -> afficher("  café : la machine coule le café"))
            .thenCompose(rien -> delai(500))
            .thenRun(() -> afficher("  café : terminé"));
}

// une opération qui rend un entier : elle rend un CompletableFuture<Integer>
CompletableFuture<Integer> compterSucres(int nombre) {
    // une chaîne d'étapes qui commence par un CompletableFuture déjà terminé
    CompletableFuture<Void> etapes = CompletableFuture.completedFuture(null);
    for (int i = 1; i <= nombre; i++) {
        // une lambda ne peut utiliser qu'une variable locale effectivement finale
        int numero = i;
        etapes = etapes.thenCompose(rien -> delai(200))
                .thenRun(() -> afficher("  sucre n° " + numero));
    }
    // à la fin de la chaîne, le résultat : le nombre de sucres
    return etapes.thenApply(rien -> nombre);
}

// l'équivalent asynchrone de Thread.sleep : un CompletableFuture terminé au bout de [ms] millisecondes
CompletableFuture<Void> delai(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// affichage avec le temps écoulé depuis le début
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async04.java
    3 ms : appel de preparerCafe
   12 ms :   café : la machine chauffe
   15 ms : preparerCafe a rendu la main : le programme continue
   16 ms : appel de compterSucres
   17 ms : compterSucres a rendu la main
  217 ms :   sucre n° 1
  417 ms :   sucre n° 2
  618 ms :   sucre n° 3
 1014 ms :   café : la machine coule le café
 1515 ms :   café : terminé
 1516 ms : café prêt, avec 3 sucre(s)

Commentaires :

• lignes 40-43 : la méthode [delai] est l'équivalent asynchrone de [Thread.sleep] : elle rend un [CompletableFuture<Void>] qui sera terminé au bout du délai, sans occuper de thread. Nous l'utiliserons dans beaucoup d'exemples pour simuler une E/S ;
• lignes 17-24 : une opération sans résultat. La méthode affiche un premier message, puis décrit l'enchaînement des étapes et rend aussitôt le [CompletableFuture] final : c'est pourquoi la ligne 4 des résultats apparaît juste après la ligne 3, et non une seconde et demie plus tard. Les étapes s'exécuteront plus tard, chacune lorsque la précédente sera terminée :
  • ligne 21 : [thenRun(action)] exécute une action quand l'opération précédente est terminée ;
  • ligne 22 : [thenCompose(fonction)] enchaîne une autre opération asynchrone : la fonction rend elle-même un [CompletableFuture] (ici un nouveau délai de 500 ms), et le [CompletableFuture] rendu par [thenCompose] ne sera terminé que lorsque cette seconde opération le sera. Avec [thenApply], on aurait obtenu un [CompletableFuture<CompletableFuture<Void>>], terminé dès le début du second délai : [thenCompose] est à [thenApply] ce que [flatMap] est à [map] dans l'API Stream ;
• lignes 27-38 : une opération qui rend un entier. Une boucle ne peut pas « attendre » à chaque tour : elle construit la chaîne des étapes, chaque tour ajoutant un délai et un affichage à la chaîne existante, en partant d'un [CompletableFuture] déjà terminé (ligne 29). Toute la chaîne est construite en quelques microsecondes, puis se déroule ensuite au fil des délais. La variable [numero] (ligne 32) est nécessaire car une lambda ne peut utiliser qu'une variable locale effectivement finale, ce que n'est pas [i]. À la fin, [thenApply] transforme la chaîne en un [CompletableFuture<Integer>] dont le résultat est le nombre de sucres ;
• lignes 6-10 : le programme principal appelle les deux méthodes. Il récupère deux [CompletableFuture] et continue son exécution. Les deux opérations progressent ensuite « en même temps » : les sucres sont comptés (lignes 7-9 des résultats) pendant que la machine chauffe ;
• ligne 12 : [preparation.thenCombine(comptage, fonction)] rend un [CompletableFuture] terminé lorsque les deux opérations le sont, avec le résultat de la fonction appliquée à leurs deux résultats : ici on garde le nombre de sucres. [join()] attend ce résultat. Le café est prêt au bout de 1,5 seconde, la durée de la plus longue des deux opérations, et non au bout de 2,1 secondes (1,5 + 3 × 0,2).

On voit les qualités et les limites de cette façon de faire. Aucun thread n'attend : les étapes sont exécutées par le thread de la minuterie, au fil des délais. Mais le code ne se lit plus comme une suite d'instructions : la boucle de [compterSucres] ne fait pas ce qu'une boucle fait d'habitude (elle construit une chaîne au lieu d'exécuter des tours), et un simple [if] ou un [try / catch] au milieu des étapes demande des méthodes dédiées. C'est le prix à payer quand le langage n'offre pas de mots clés comme [await].

9.3.2. Seconde façon : du code bloquant sur des threads virtuels

Les threads virtuels permettent d'écrire les mêmes méthodes comme des méthodes ordinaires, bloquantes. Le programme [Async05.java] prépare le même café :

// le même café avec des threads virtuels : des méthodes ordinaires, bloquantes, qui se lisent de haut en bas
final long debut = System.nanoTime();

void main() throws InterruptedException, ExecutionException {
    // un exécuteur qui crée un thread virtuel pour chaque tâche qu'on lui confie
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        afficher("lancement de preparerCafe");
        Future<?> preparation = executeur.submit(this::preparerCafe);
        afficher("preparerCafe est lancé : le programme continue");
        afficher("lancement de compterSucres");
        Future<Integer> comptage = executeur.submit(() -> compterSucres(3));
        afficher("compterSucres est lancé");
        // on attend maintenant les deux résultats
        preparation.get();
        int nbSucres = comptage.get();
        afficher("café prêt, avec " + nbSucres + " sucre(s)");
    }
}

// une méthode ordinaire qui ne rend pas de résultat
void preparerCafe() {
    afficher("  café : la machine chauffe");
    // une attente bloquante : sur un thread virtuel, elle libère le thread porteur
    dormir(1000);
    afficher("  café : la machine coule le café");
    dormir(500);
    afficher("  café : terminé");
}

// une méthode ordinaire qui rend un entier
int compterSucres(int nombre) {
    for (int i = 1; i <= nombre; i++) {
        dormir(200);
        afficher("  sucre n° " + i);
    }
    return nombre;
}

// Thread.sleep sans exception contrôlée : une interruption devient une CancellationException
void dormir(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        // on rétablit l'indicateur d'interruption du thread avant de lancer l'exception
        Thread.currentThread().interrupt();
        throw new CancellationException("attente interrompue");
    }
}

// affichage avec le temps écoulé depuis le début
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async05.java
    3 ms : lancement de preparerCafe
   17 ms :   café : la machine chauffe
   18 ms : preparerCafe est lancé : le programme continue
   18 ms : lancement de compterSucres
   19 ms : compterSucres est lancé
  221 ms :   sucre n° 1
  422 ms :   sucre n° 2
  623 ms :   sucre n° 3
 1020 ms :   café : la machine coule le café
 1521 ms :   café : terminé
 1522 ms : café prêt, avec 3 sucre(s)

Commentaires :

• lignes 20-37 : les deux opérations sont des méthodes ordinaires : [preparerCafe] ne rend rien ([void]), [compterSucres] rend un [int]. Les attentes sont des [Thread.sleep] bloquants, la boucle est une vraie boucle. Rien, dans ces méthodes, n'indique qu'elles seront exécutées de façon concurrente : on pourrait les appeler directement ;
• lignes 39-48 : la méthode [dormir] appelle [Thread.sleep] et transforme une éventuelle [InterruptedException] (exception contrôlée) en [CancellationException] (non contrôlée), après avoir rétabli l'indicateur d'interruption du thread : c'est la règle quand on intercepte une [InterruptedException] sans la propager (nous y reviendrons au paragraphe sur l'annulation) ;
• ligne 6 : un exécuteur qui crée un thread virtuel pour chaque tâche. Il est déclaré dans un [try] avec ressource : sa méthode [close()] attendra la fin de toutes les tâches ;
• lignes 8 et 11 : [submit] confie chaque méthode à un nouveau thread virtuel et rend un [Future]. La référence de méthode [this::preparerCafe] est un [Runnable] (la méthode ne rend rien), d'où un [Future<?>] ; la lambda [() -> compterSucres(3)] est un [Callable<Integer>], d'où un [Future<Integer>] ;
• lignes 14-15 : le programme principal attend les deux résultats avec [get()]. C'est une attente bloquante du thread principal, sans conséquence ici ;
• les résultats sont les mêmes qu'avec [Async04.java] : 1,5 seconde au total, les sucres comptés pendant que la machine chauffe. Pendant les attentes, aucun thread du système n'est bloqué : les deux threads virtuels sont décrochés de leurs threads porteurs.

Avec les threads virtuels, l'asynchronisme n'est plus une affaire d'écriture du code mais d'exécution : on écrit du code bloquant, et c'est le choix de l'exécuteur (threads virtuels plutôt que pool de threads de plate-forme) qui décide si les attentes immobilisent des threads du système. C'est pourquoi les bibliothèques Java existantes (JDBC, fichiers, sockets, [HttpClient.send]…), toutes écrites de façon bloquante, profitent des threads virtuels sans modification.

9.3.3. Ce que fait la JVM

Comment un thread virtuel peut-il s'interrompre au milieu d'une boucle, libérer son thread porteur, puis reprendre plus tard au même endroit, avec les mêmes valeurs de variables locales, éventuellement sur un autre thread porteur ? Un thread virtuel est un objet Java ([java.lang.VirtualThread], une classe dérivée de [Thread]) qui possède sa propre pile d'appels. Pour l'exécuter, la JVM le monte (« mount ») sur un thread porteur, un thread de plate-forme d'un [ForkJoinPool] réservé à cet usage, qui compte par défaut autant de threads que de processeurs. Lorsque le thread virtuel appelle une opération bloquante ([Thread.sleep], la lecture d'un socket, [BlockingQueue.take], l'attente d'un verrou…), les classes de la JVM, réécrites pour cela en Java 21, ne bloquent pas le thread porteur :

• elles enregistrent l'opération auprès du système (une minuterie pour un délai, le mécanisme de notification d'E/S du système, [epoll] sous Linux, pour un socket) ;
• elles démontent (« unmount ») le thread virtuel : les cadres de sa pile d'appels sont copiés dans le tas, sous la forme d'un objet appelé continuation, et le thread porteur est rendu au [ForkJoinPool], qui lui confie un autre thread virtuel prêt à s'exécuter ;
• lorsque l'opération se termine, le thread virtuel redevient prêt : il sera remonté sur le premier thread porteur libre, ses cadres de pile seront recopiés, et l'exécution reprendra juste après l'appel bloquant.

C'est le principe des méthodes de rappel (« rappelle-moi quand l'opération sera terminée »), mais c'est la JVM qui découpe l'exécution, sauvegarde les variables et rétablit le contexte, au niveau de la pile d'appels, sans que le code source en porte la trace. D'autres langages confient ce travail au compilateur, qui transforme chaque méthode [async] en une « machine à états » ; Java le confie à la JVM, ce qui lui permet de traiter de la même façon tout le code existant.

Deux limites sont à connaître :

• certaines opérations ne peuvent pas démonter le thread virtuel ; il reste alors épinglé (« pinned ») sur son thread porteur, qui est bloqué pendant l'attente. C'est le cas pendant l'exécution d'une méthode native, et de la plupart des lectures et écritures de fichiers (le système ne propose pas, pour les fichiers, de notification utilisable par la JVM) ; pour ces dernières, le [ForkJoinPool] compense en créant temporairement un thread porteur supplémentaire. Jusqu'à Java 23, c'était aussi le cas d'un thread virtuel qui attendait à l'intérieur d'un bloc [synchronized] ; depuis Java 24, ce n'est plus le cas ;
• un thread virtuel n'accélère pas un calcul : un thread virtuel qui calcule occupe son thread porteur tout le temps du calcul. Les threads virtuels servent les programmes qui attendent beaucoup, pas ceux qui calculent beaucoup (pour ces derniers, on dispose des pools de threads et des flux parallèles du chapitre précédent).

9.3.4. Quel thread exécute la suite ?

Le programme [Async06.java] affiche le thread qui exécute chaque morceau de code :

// quel thread exécute la suite d'un traitement ?
void main() throws InterruptedException {
    afficher("début du programme, parallélisme du pool commun : " + ForkJoinPool.getCommonPoolParallelism());

    // 1 - la continuation d'un CompletableFuture déjà terminé est exécutée tout de suite, par l'appelant
    CompletableFuture.completedFuture(42).thenAccept(valeur -> afficher("1 - thenAccept, résultat connu"));

    // 2 - sinon, elle est exécutée par le thread qui termine le CompletableFuture
    delai(100).thenRun(() -> afficher("2 - thenRun après un délai")).join();
    CompletableFuture<String> commande = new CompletableFuture<>();
    commande.thenAccept(texte -> afficher("2 - thenAccept après complete(\"" + texte + "\")"));
    Thread.ofPlatform().name("livreur").start(() -> commande.complete("livrée")).join();

    // 3 - les variantes xxxAsync confient la continuation à un exécuteur
    delai(100).thenRunAsync(() -> afficher("3 - thenRunAsync, exécuteur par défaut")).join();
    ThreadFactory fabrique = Thread.ofPlatform().name("pool-", 1).factory();
    try (ExecutorService pool = Executors.newFixedThreadPool(2, fabrique)) {
        delai(100).thenRunAsync(() -> afficher("3 - thenRunAsync, exécuteur choisi"), pool).join();
        CompletableFuture.runAsync(() -> afficher("3 - runAsync, exécuteur choisi"), pool).join();
    }
    try (ExecutorService virtuels = Executors.newVirtualThreadPerTaskExecutor()) {
        CompletableFuture.runAsync(() -> afficher("3 - runAsync, threads virtuels"), virtuels).join();
    }

    // 4 - un thread virtuel peut changer de thread porteur après chaque attente
    Thread a = Thread.ofVirtual().start(() -> travailler("A"));
    Thread b = Thread.ofVirtual().start(() -> travailler("B"));
    a.join();
    b.join();
    afficher("fin du programme");
}

void travailler(String nom) {
    for (int i = 1; i <= 3; i++) {
        afficher("  " + nom + i + " : avant l'attente");
        dormir(100);
    }
}

// un CompletableFuture terminé au bout de [ms] millisecondes
CompletableFuture<Void> delai(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// Thread.sleep sans exception contrôlée : une interruption devient une CancellationException
void dormir(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("attente interrompue");
    }
}

// affichage préfixé par le thread qui l'exécute
// (nom d'un thread de plate-forme, description d'un thread virtuel)
void afficher(String message) {
    Thread thread = Thread.currentThread();
    IO.println("[" + (thread.isVirtual() ? thread.toString() : thread.getName()) + "] " + message);
}

Résultats :

> java Async06.java
[main] début du programme, parallélisme du pool commun : 1
[main] 1 - thenAccept, résultat connu
[ForkJoinPool.commonPool-delayScheduler] 2 - thenRun après un délai
[livreur] 2 - thenAccept après complete("livrée")
[ForkJoinPool.commonPool-worker-1] 3 - thenRunAsync, exécuteur par défaut
[pool-1] 3 - thenRunAsync, exécuteur choisi
[pool-2] 3 - runAsync, exécuteur choisi
[VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1] 3 - runAsync, threads virtuels
[VirtualThread[#31]/runnable@ForkJoinPool-1-worker-1]   A1 : avant l'attente
[VirtualThread[#32]/runnable@ForkJoinPool-1-worker-2]   B1 : avant l'attente
[VirtualThread[#32]/runnable@ForkJoinPool-1-worker-2]   B2 : avant l'attente
[VirtualThread[#31]/runnable@ForkJoinPool-1-worker-1]   A2 : avant l'attente
[VirtualThread[#31]/runnable@ForkJoinPool-1-worker-2]   A3 : avant l'attente
[VirtualThread[#32]/runnable@ForkJoinPool-1-worker-1]   B3 : avant l'attente
[main] fin du programme

Commentaires :

• ligne 3 : le pool commun ([ForkJoinPool.commonPool()]), l'exécuteur par défaut des [CompletableFuture], a un parallélisme égal au nombre de processeurs moins un, soit 1 sur notre machine à 2 processeurs (ligne 2 des résultats) ;
• ligne 6 : la continuation d'un [CompletableFuture] déjà terminé est exécutée immédiatement, par le thread qui l'attache : le thread [main] (ligne 3) ;
• lignes 9-12 : si l'opération n'est pas terminée, la continuation sera exécutée par le thread qui la terminera : le thread de la minuterie des délais pour [delai(100)] (ligne 4), le thread [livreur] qui appelle [complete] pour la commande (ligne 5). On ne choisit donc pas le thread qui exécute une continuation non asynchrone ; c'est sans importance pour une action brève, mais une continuation longue ou bloquante immobiliserait ce thread (celui de la minuterie, par exemple, qui sert tous les délais des [CompletableFuture]) ;
• lignes 15-23 : les variantes [xxxAsync] ([thenRunAsync], [thenApplyAsync]…) confient la continuation à un exécuteur : sans paramètre supplémentaire, au pool commun (ligne 6 des résultats) ; avec un exécuteur en dernier paramètre, à cet exécuteur. On peut ainsi choisir des threads de plate-forme (lignes 7-8 : un pool de deux threads dont les noms sont fixés par une fabrique de threads, [ThreadFactory], construite par [Thread.ofPlatform().name("pool-", 1).factory()]), ou des threads virtuels (ligne 9). [supplyAsync] et [runAsync] acceptent de même un exécuteur ;
• lignes 26-38 : deux threads virtuels qui affichent un message puis dorment, trois fois de suite. Les résultats montrent qu'ils ne restent pas attachés à un thread porteur : le thread virtuel n° 31 (A) commence sur [worker-1] et termine sur [worker-2], le n° 32 (B) fait l'inverse. Pendant que l'un dort, son porteur est libre pour l'autre. On ne doit donc jamais supposer, dans un thread virtuel, que le code situé après une attente est exécuté par le même thread porteur que le code situé avant ; mais c'est toujours le même thread virtuel, et c'est lui qui compte pour le programme ([Thread.currentThread()] rend le thread virtuel, ses variables [ThreadLocal] sont conservées).

Cette façon de mener plusieurs opérations de front sans multiplier les threads du système s'appelle la concurrence (« concurrency »). Elle se distingue du parallélisme, qui consiste à exécuter plusieurs calculs au même instant sur plusieurs processeurs (c'est l'affaire du chapitre précédent : pools de threads, flux parallèles). L'asynchronisme sert la concurrence des E/S ; il ne rend pas les calculs plus rapides.

Le choix de l'exécuteur est donc important :

• le pool commun convient aux continuations brèves et aux calculs courts ; il ne faut pas y exécuter d'opération bloquante, au risque de la famine décrite avec [Async01.java] (il n'a qu'un thread sur notre machine) ;
• pour une opération bloquante (E/S, appel JDBC, [Thread.sleep]), on passe un exécuteur de threads virtuels : [CompletableFuture.supplyAsync(fournisseur, executeurVirtuel)] ;
• pour une interface graphique, on passe un exécuteur qui fait exécuter la continuation par le thread de l'interface, ce qui permet de mettre à jour les composants graphiques : avec Swing, [thenAcceptAsync(action, SwingUtilities::invokeLater)] (la référence de méthode [SwingUtilities::invokeLater] est un [Executor] valable, puisque [Executor] est une interface fonctionnelle dont l'unique méthode reçoit un [Runnable]) ; avec JavaFX, [Platform::runLater].

9.3.5. Le programme principal

Dans d'autres langages, la méthode principale d'un programme asynchrone doit être déclarée de façon particulière. En Java, il n'y a rien de tel : la méthode [main] est exécutée par le thread de plate-forme [main], et c'est à elle d'attendre la fin des opérations qu'elle a lancées, avec [join()] ou [get()]. C'est le seul endroit où une attente bloquante sur un [CompletableFuture] est normale. Le programme [Async07.java] réalise un compte à rebours asynchrone :

// le programme principal attend la fin des opérations asynchrones et fixe le code de sortie
void main(String[] args) {
    // le nombre de secondes est passé en argument
    if (args.length != 1 || !args[0].matches("\\d+")) {
        IO.println("syntaxe : java Async07.java nbSecondes");
        // le code de sortie du programme : 0 pour un succès, autre chose pour une erreur
        System.exit(1);
    }
    CompletableFuture<Void> compteARebours = compterARebours(Integer.parseInt(args[0]));
    IO.println("compte à rebours lancé");
    // sans cette attente, main se terminerait aussitôt et le programme avec lui
    compteARebours.join();
    IO.println("terminé");
}

// une boucle asynchrone s'écrit par récursivité : chaque étape lance la suivante
CompletableFuture<Void> compterARebours(int nbSecondes) {
    if (nbSecondes == 0) {
        return CompletableFuture.completedFuture(null);
    }
    IO.println("encore " + nbSecondes + " s");
    return delai(1000).thenCompose(rien -> compterARebours(nbSecondes - 1));
}

// un CompletableFuture terminé au bout de [ms] millisecondes
CompletableFuture<Void> delai(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

Résultats, avec puis sans argument :

> java Async07.java 3
encore 3 s
compte à rebours lancé
encore 2 s
encore 1 s
terminé

> java Async07.java
syntaxe : java Async07.java nbSecondes
> echo $?
1

Commentaires :

• ligne 2 : le paramètre [args] de [main] reçoit les arguments de la ligne de commande, placés après le nom du fichier : [java Async07.java 3] ;
• lignes 4-8 : si l'argument est absent ou n'est pas un entier positif (expression régulière [\d+] : un ou plusieurs chiffres), le programme s'arrête avec [System.exit(1)]. L'entier passé à [System.exit] est le code de sortie du programme (0 pour un succès, autre chose pour une erreur), que peut tester un script : [echo $?] sous Linux (second résultat), [echo %ERRORLEVEL%] sous Windows. Un programme dont la méthode [main] se termine normalement sort avec le code 0 ;
• lignes 16-23 : un compte à rebours asynchrone. Une boucle asynchrone s'écrit naturellement par récursivité : chaque étape affiche un message puis enchaîne, avec [thenCompose], un délai d'une seconde suivi de l'étape suivante. Il n'y a pas d'empilement d'appels : l'appel récursif est fait par la continuation, une seconde plus tard ;
• lignes 9-12 : la méthode [compterARebours] affiche son premier message puis rend la main (lignes 2-3 des résultats) ; le programme principal attend la fin de l'opération avec [join()]. Sans cette attente, [main] se terminerait aussitôt et le programme avec elle : la JVM s'arrête lorsqu'il ne reste plus que des threads démons (« daemon threads »), et les threads de la minuterie et du pool commun sont des démons. Il en est de même des threads virtuels, qui sont toujours des démons : un programme doit attendre la fin des threads virtuels qu'il a lancés (avec [join], [get] ou la méthode [close] de leur exécuteur).

9.4. Exécution séquentielle et exécution concurrente

9.4.1. allOf et invokeAll

L'exemple du café l'a montré : le moment où l'on lance une opération asynchrone et le moment où on l'attend peuvent être différents. Écrire [methode().join()] lance l'opération et l'attend aussitôt ; écrire [var f = methode();] puis plus tard [f.join();] permet de lancer d'autres opérations entre les deux. Le programme [Async08.java] compare les deux façons de faire sur des opérations d'une seconde :

// exécution séquentielle et exécution concurrente d'opérations d'une seconde
long debut = System.nanoTime();

void main() throws InterruptedException {
    // 1 - en séquence : chaque opération est attendue avant de lancer la suivante
    int a = lireTemperature("Angers").join();
    int b = lireTemperature("Brest").join();
    int c = lireTemperature("Colmar").join();
    IO.println("en séquence : " + a + ", " + b + ", " + c + " en " + ecoule() + " ms");

    // 2 - en concurrence : on lance les trois opérations, puis on les attend
    debut = System.nanoTime();
    CompletableFuture<Integer> fa = lireTemperature("Angers");
    CompletableFuture<Integer> fb = lireTemperature("Brest");
    CompletableFuture<Integer> fc = lireTemperature("Colmar");
    CompletableFuture.allOf(fa, fb, fc).join();
    IO.println("en concurrence : " + fa.resultNow() + ", " + fb.resultNow() + ", " + fc.resultNow()
            + " en " + ecoule() + " ms");

    // 3 - allOf sur une liste de CompletableFuture, puis la liste des résultats
    debut = System.nanoTime();
    List<String> villes = List.of("Angers", "Brest", "Colmar", "Dijon", "Évry", "Foix");
    List<CompletableFuture<Integer>> lectures = villes.stream().map(this::lireTemperature).toList();
    List<Integer> temperatures = CompletableFuture.allOf(lectures.toArray(CompletableFuture[]::new))
            .thenApply(rien -> lectures.stream().map(CompletableFuture::join).toList())
            .join();
    IO.println(villes.size() + " villes : " + temperatures + " en " + ecoule() + " ms");

    // 4 - la même chose avec des threads virtuels et une méthode bloquante
    debut = System.nanoTime();
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        List<Callable<Integer>> taches = villes.stream()
                .map(ville -> (Callable<Integer>) () -> lireTemperatureBloquant(ville))
                .toList();
        // invokeAll lance toutes les tâches et rend leurs Future quand elles sont toutes terminées
        List<Integer> resultats = executeur.invokeAll(taches).stream().map(Future::resultNow).toList();
        IO.println(villes.size() + " villes (threads virtuels) : " + resultats + " en " + ecoule() + " ms");
    }
}

// simule l'interrogation d'un service distant qui répond en une seconde
CompletableFuture<Integer> lireTemperature(String ville) {
    // une « température » déduite du nom de la ville
    return delai(1000).thenApply(rien -> 10 + ville.length());
}

// la même opération, écrite de façon bloquante
int lireTemperatureBloquant(String ville) throws InterruptedException {
    Thread.sleep(1000);
    return 10 + ville.length();
}

// un CompletableFuture terminé au bout de [ms] millisecondes
CompletableFuture<Void> delai(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// millisecondes écoulées depuis [debut]
long ecoule() {
    return (System.nanoTime() - debut) / 1_000_000;
}

Résultats :

> java Async08.java
en séquence : 16, 15, 16 en 3007 ms
en concurrence : 16, 15, 16 en 1000 ms
6 villes : [16, 15, 16, 15, 14, 14] en 1006 ms
6 villes (threads virtuels) : [16, 15, 16, 15, 14, 14] en 1013 ms

Commentaires :

• lignes 41-45 : l'opération simulée : un délai d'une seconde, puis une « température » déduite de la longueur du nom de la ville ;
• lignes 6-8 : chaque opération est lancée puis attendue avant de lancer la suivante. Les trois opérations se font l'une après l'autre : 3 secondes (ligne 2 des résultats). C'est la bonne façon de faire lorsqu'une opération a besoin du résultat de la précédente (on écrirait alors plutôt une chaîne de [thenCompose]) ;
• lignes 13-15 : les trois opérations sont lancées sans être attendues. Elles se déroulent en même temps ;
• ligne 16 : [CompletableFuture.allOf(...)] rend un [CompletableFuture<Void>] qui se termine lorsque toutes les opérations qui lui sont passées sont terminées. On l'attend avec [join()] : 1 seconde au lieu de 3 (ligne 3) ;
• lignes 17-18 : les trois opérations sont terminées, [resultNow()] rend donc leur résultat sans attendre ;
• lignes 22-27 : le cas d'une liste d'opérations. [allOf] attend un tableau ([CompletableFuture<?>...]) : la méthode [toArray(CompletableFuture[]::new)] de la liste le fabrique. [allOf] ne rend pas les résultats : on les recueille dans une continuation, en appelant [join()] sur chaque opération, ce qui ne bloque pas puisqu'elles sont toutes terminées. On obtient la liste des résultats dans l'ordre des opérations (et non dans l'ordre où elles se sont terminées). L'expression [villes.stream().map(this::lireTemperature).toList()] appelle la méthode pour chaque ville et produit ainsi six opérations lancées en même temps ; les six durent ensemble une seconde (ligne 4) ;
• lignes 29-38 : la même chose avec une méthode bloquante (lignes 47-51) exécutée par des threads virtuels. La méthode [invokeAll] d'un exécuteur reçoit une collection de [Callable<T>], les lance toutes et rend la liste de leurs [Future<T>] lorsqu'elles sont toutes terminées : c'est l'équivalent bloquant de [allOf]. Le transtypage [(Callable<Integer>)] de la ligne 33 indique au compilateur le type de la lambda, qu'il ne peut pas déduire seul dans un [map]. Les six opérations durent ensemble une seconde (ligne 5).

9.4.2. anyOf

[CompletableFuture.anyOf(...)] rend un [CompletableFuture<Object>] qui se termine dès que l'une des opérations qui lui sont passées est terminée, avec le résultat (ou l'exception) de celle-ci. Le programme [Async09.java] l'utilise de deux façons :

// anyOf : réagir à la première opération terminée
long debut = System.nanoTime();

void main() {
    // 1 - on interroge trois serveurs miroirs et on garde la première réponse
    List<CompletableFuture<String>> requetes = List.of(
            interroger("miroir-paris", 800),
            interroger("miroir-lyon", 300),
            interroger("miroir-lille", 600));
    // anyOf rend un CompletableFuture<Object> qui se termine comme la première opération terminée
    Object premiere = CompletableFuture.anyOf(requetes.toArray(CompletableFuture[]::new)).join();
    afficher("première réponse : " + premiere);
    // les autres requêtes continuent : on les attend ici pour voir leurs réponses
    CompletableFuture.allOf(requetes.toArray(CompletableFuture[]::new)).join();
    afficher("toutes les réponses : " + requetes.stream().map(CompletableFuture::resultNow).toList());

    // 2 - une opération limitée dans le temps : on fait « la course » avec un délai
    debut = System.nanoTime();
    CompletableFuture<String> lente = interroger("serveur-lent", 2000);
    CompletableFuture<Object> course = CompletableFuture.anyOf(lente, delai(500));
    course.join();
    if (lente.isDone()) {
        afficher("réponse obtenue à temps : " + lente.join());
    } else {
        afficher("le serveur lent n'a pas répondu dans le délai de 500 ms");
    }
}

// simule un serveur qui répond au bout de [duree] millisecondes
CompletableFuture<String> interroger(String serveur, int duree) {
    return delai(duree).thenApply(rien -> serveur + " (" + duree + " ms)");
}

// un CompletableFuture terminé au bout de [ms] millisecondes
CompletableFuture<Void> delai(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// affichage avec le temps écoulé depuis [debut]
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async09.java
  309 ms : première réponse : miroir-lyon (300 ms)
  806 ms : toutes les réponses : [miroir-paris (800 ms), miroir-lyon (300 ms), miroir-lille (600 ms)]
  500 ms : le serveur lent n'a pas répondu dans le délai de 500 ms

Commentaires :

• lignes 6-9 : on interroge trois serveurs miroirs qui détiennent la même information et répondent plus ou moins vite. Les trois opérations sont rangées dans une liste ;
• ligne 11 : [anyOf] rend un [CompletableFuture<Object>] dont le résultat est celui de la première opération terminée. Son type est [Object] car les opérations passées à [anyOf] peuvent avoir des types de résultats différents : il faudrait le transtyper pour l'utiliser comme une chaîne. La première réponse arrive au bout de 300 ms (ligne 2 des résultats) ;
• ligne 14 : les autres requêtes ne sont pas arrêtées par [anyOf] : elles continuent jusqu'à leur fin. Ici on les attend pour afficher toutes les réponses (ligne 3) ;
• lignes 19-26 : un usage classique de [anyOf] : limiter la durée d'une attente en faisant « la course » entre l'opération et un délai. Si l'opération n'est pas terminée quand la course l'est, c'est le délai qui a gagné : l'opération a pris trop de temps (ligne 4). [CompletableFuture] propose une écriture plus directe de ce cas, la méthode [orTimeout], présentée plus loin.

9.4.3. Traiter les résultats au fur et à mesure

On veut souvent traiter les résultats de plusieurs opérations au fur et à mesure qu'elles se terminent, sans attendre la plus lente. Avec des [CompletableFuture], c'est naturel : il suffit d'attacher une continuation à chaque opération. Avec des tâches soumises à un exécuteur, la classe [ExecutorCompletionService] (Java 5) livre les [Future] dans l'ordre où les tâches se terminent. Le programme [Async10.java] traite cinq « fichiers » dont le traitement dure de 100 à 900 ms, des deux façons :

// traiter les résultats dans l'ordre où les opérations se terminent
long debut = System.nanoTime();
final int[] durees = {900, 300, 700, 100, 500};

void main() throws InterruptedException, ExecutionException {
    // 1 - avec des CompletableFuture : une continuation par opération, exécutée dès qu'elle se termine
    List<CompletableFuture<Void>> affichages = new ArrayList<>();
    for (int i = 0; i < durees.length; i++) {
        affichages.add(traiter("fichier" + i + ".txt", durees[i]).thenAccept(this::afficher));
    }
    CompletableFuture.allOf(affichages.toArray(CompletableFuture[]::new)).join();
    afficher("1 - tous les fichiers ont été traités");

    // 2 - avec des threads virtuels : un CompletionService livre les Future dans l'ordre de leur fin
    debut = System.nanoTime();
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        CompletionService<String> service = new ExecutorCompletionService<>(executeur);
        for (int i = 0; i < durees.length; i++) {
            String fichier = "fichier" + i + ".txt";
            int duree = durees[i];
            service.submit(() -> traiterBloquant(fichier, duree));
        }
        // take() attend (en bloquant) la prochaine tâche terminée
        for (int i = 0; i < durees.length; i++) {
            afficher(service.take().get());
        }
    }
    afficher("2 - tous les fichiers ont été traités");
}

// simule le traitement d'un fichier qui dure [duree] millisecondes
CompletableFuture<String> traiter(String fichier, int duree) {
    return new CompletableFuture<String>()
            .completeOnTimeout(fichier + " traité en " + duree + " ms", duree, TimeUnit.MILLISECONDS);
}

// le même traitement, écrit de façon bloquante
String traiterBloquant(String fichier, int duree) throws InterruptedException {
    Thread.sleep(duree);
    return fichier + " traité en " + duree + " ms";
}

// affichage avec le temps écoulé depuis [debut]
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async10.java
  107 ms : fichier3.txt traité en 100 ms
  318 ms : fichier1.txt traité en 300 ms
  508 ms : fichier4.txt traité en 500 ms
  707 ms : fichier2.txt traité en 700 ms
  906 ms : fichier0.txt traité en 900 ms
  907 ms : 1 - tous les fichiers ont été traités
  115 ms : fichier3.txt traité en 100 ms
  314 ms : fichier1.txt traité en 300 ms
  514 ms : fichier4.txt traité en 500 ms
  714 ms : fichier2.txt traité en 700 ms
  914 ms : fichier0.txt traité en 900 ms
  915 ms : 2 - tous les fichiers ont été traités

Commentaires :

• lignes 7-11 : chaque traitement reçoit sa propre continuation, [thenAccept(this::afficher)], exécutée dès qu'il se termine : le fichier 3 (100 ms) est affiché en premier, le fichier 0 (900 ms) en dernier (lignes 2-6 des résultats). La méthode [traiter] (lignes 31-35) rend un [CompletableFuture<String>] terminé au bout de la durée indiquée avec le message voulu ([completeOnTimeout] reçoit directement la valeur du résultat). La liste des continuations sert à attendre la fin de tous les affichages avec [allOf] ;
• lignes 16-27 : un [CompletionService<T>] enveloppe un exécuteur (ici un exécuteur de threads virtuels). Sa méthode [submit] lance une tâche ; sa méthode [take()] attend, en bloquant, la prochaine tâche terminée et rend son [Future], dont [get()] rend aussitôt le résultat. La boucle des lignes 24-26 traite donc les fichiers dans l'ordre où ils sont terminés (lignes 8-12) ;
• dans les deux cas, l'ensemble dure 900 ms, la durée du plus long traitement, et chaque résultat est affiché au plus tôt.

9.5. Les exceptions dans le code asynchrone

9.5.1. Propagation, exceptionally, handle, whenComplete

Une exception lancée par une opération asynchrone n'est pas lancée vers le code qui a lancé l'opération : elle est rangée dans le [CompletableFuture], qui passe à l'état [FAILED]. Elle se propage ensuite le long de la chaîne des continuations, jusqu'à une méthode qui la traite, ou jusqu'à [join()] qui la relance enveloppée dans une [CompletionException]. Le programme [Async11.java] examine les différents cas :

// les exceptions dans le code asynchrone
void main() {
    // 1 - join() lance une CompletionException qui enveloppe l'exception de l'opération
    try {
        int valeur = lireValeur("douze").join();
        IO.println("valeur = " + valeur);
    } catch (CompletionException e) {
        IO.println("1 - join() lance " + nom(e) + ", de cause " + nom(e.getCause())
                + " : " + e.getCause().getMessage());
    }

    // 2 - exceptionally fournit une valeur de remplacement en cas d'échec
    int valeur = lireValeur("treize").exceptionally(e -> -1).join();
    IO.println("2 - exceptionally : valeur de remplacement " + valeur);

    // 3 - handle reçoit le résultat ou l'exception ; whenComplete les observe sans rien changer
    for (String texte : List.of("14", "quinze")) {
        String bilan = lireValeur(texte)
                .whenComplete((v, e) -> IO.println("3 - whenComplete a vu : " + (e == null ? v : nom(e))))
                .handle((v, e) -> e == null ? "valeur " + v : "échec, " + nom(cause(e)))
                .join();
        IO.println("3 - handle a rendu : " + bilan);
    }

    // 4 - une méthode qui rend un CompletableFuture peut aussi lancer une exception dès l'appel
    try {
        lireValeur(" ");
        IO.println("4 - l'appel n'a rien lancé");
    } catch (IllegalArgumentException e) {
        IO.println("4 - c'est l'appel lui-même qui lance " + nom(e) + " : " + e.getMessage());
    }

    // 5 - allOf avec plusieurs opérations en échec
    List<CompletableFuture<Integer>> lectures = Stream.of("1", "deux", "3", "quatre")
            .map(this::lireValeur)
            .toList();
    try {
        CompletableFuture.allOf(lectures.toArray(CompletableFuture[]::new)).join();
    } catch (CompletionException e) {
        // join() ne relance qu'une des exceptions…
        IO.println("5 - allOf échoue avec " + nom(e.getCause()) + " : " + e.getCause().getMessage());
        // … mais chaque opération a conservé son exception ou son résultat
        for (CompletableFuture<Integer> lecture : lectures) {
            switch (lecture.state()) {
                case SUCCESS -> IO.println("      réussite, résultat = " + lecture.resultNow());
                case FAILED -> IO.println("      échec, " + nom(lecture.exceptionNow())
                        + " : " + lecture.exceptionNow().getMessage());
                default -> IO.println("      état inattendu : " + lecture.state());
            }
        }
    }
}

// une opération asynchrone qui échoue si la chaîne n'est pas un entier
CompletableFuture<Integer> lireValeur(String texte) {
    // vérification immédiate du paramètre, avant de lancer l'opération
    if (texte == null || texte.isBlank()) {
        throw new IllegalArgumentException("le texte ne doit pas être vide");
    }
    // la lecture prend un peu de temps
    return new CompletableFuture<Void>()
            .completeOnTimeout(null, 100, TimeUnit.MILLISECONDS)
            .thenApply(rien -> Integer.parseInt(texte));
}

// les continuations reçoivent souvent une CompletionException : on remonte à sa cause
Throwable cause(Throwable e) {
    return e instanceof CompletionException && e.getCause() != null ? e.getCause() : e;
}

// le nom simple de la classe d'une exception
String nom(Throwable e) {
    return e.getClass().getSimpleName();
}

Résultats :

> java Async11.java
1 - join() lance CompletionException, de cause NumberFormatException : For input string: "douze"
2 - exceptionally : valeur de remplacement -1
3 - whenComplete a vu : 14
3 - handle a rendu : valeur 14
3 - whenComplete a vu : CompletionException
3 - handle a rendu : échec, NumberFormatException
4 - c'est l'appel lui-même qui lance IllegalArgumentException : le texte ne doit pas être vide
5 - allOf échoue avec NumberFormatException : For input string: "deux"
      réussite, résultat = 1
      échec, NumberFormatException : For input string: "deux"
      réussite, résultat = 3
      échec, NumberFormatException : For input string: "quatre"

Commentaires :

• lignes 54-64 : la méthode testée. Elle vérifie d'abord son paramètre et lance une [IllegalArgumentException] si la chaîne est [null] ou blanche. Sinon, elle rend une opération qui attend 100 ms puis convertit la chaîne en entier, ce qui lance une [NumberFormatException] si la chaîne n'est pas un nombre ;
• lignes 4-10 : le cas ordinaire. L'exception lancée par [Integer.parseInt] dans la continuation de la ligne 63 est rangée dans le [CompletableFuture] ; [join()] la relance enveloppée dans une [CompletionException], dont [getCause()] rend l'exception d'origine (ligne 2 des résultats) ;
• ligne 13 : [exceptionally(fonction)] rend un [CompletableFuture] qui a le même résultat que l'opération si elle réussit, et sinon le résultat de la fonction appliquée à l'exception : c'est le [catch] du monde des [CompletableFuture]. Ici on remplace un échec par la valeur -1 (ligne 3) ;
• lignes 17-23 : deux autres méthodes, appliquées à une opération qui réussit puis à une opération qui échoue :
  • [whenComplete(action)] reçoit le résultat et l'exception (l'un des deux vaut [null]) et ne change rien : le [CompletableFuture] qu'elle rend a le même résultat ou la même exception que l'opération. Elle sert à observer (écrire dans un journal, libérer une ressource) : c'est le [finally] du monde des [CompletableFuture] ;
  • [handle(fonction)] reçoit aussi le résultat et l'exception, et rend une nouvelle valeur dans les deux cas : elle transforme un succès comme un échec en un résultat ;
  • les lignes 6 et 7 des résultats montrent un piège : l'exception reçue par les continuations n'est pas la [NumberFormatException] mais une [CompletionException] qui l'enveloppe, parce que l'échec s'est produit dans une étape précédente de la chaîne (la continuation de la ligne 63). La méthode [cause] (lignes 66-69) remonte à l'exception d'origine : on prendra l'habitude de l'écrire ;
• lignes 26-31 : une différence importante avec les méthodes [async] d'autres langages : une méthode qui rend un [CompletableFuture] est une méthode ordinaire, qui peut lancer une exception dès l'appel, avant d'avoir créé le moindre [CompletableFuture]. C'est le cas ici pour une chaîne blanche (ligne 8 des résultats). Le code appelant doit donc s'attendre aux deux formes d'erreur, ou la méthode doit rendre une opération déjà en échec ([return CompletableFuture.failedFuture(new IllegalArgumentException(...));]), ce qui est souvent préférable ;
• lignes 34-38 : on attend avec [allOf] quatre opérations dont deux échouent. Le [CompletableFuture] de [allOf] échoue à son tour, mais avec une seule des exceptions (ligne 9 des résultats) ;
• lignes 43-50 : chaque opération a conservé son résultat ou son exception. On les examine avec un [switch] sur leur état ; [exceptionNow()] rend l'exception d'origine, sans enveloppe. L'échec de [allOf] n'empêche pas les autres opérations de réussir (lignes 10 et 12).

Avec les threads virtuels, la question ne se pose pas dans les mêmes termes : le code exécuté par un thread virtuel est du code ordinaire, où les exceptions se propagent comme d'habitude et s'interceptent par [try / catch]. Seule la frontière entre deux threads demande une précaution : l'exception d'une tâche soumise à un exécuteur est rangée dans son [Future], et [get()] la relance enveloppée dans une [ExecutionException].

9.5.2. Les opérations oubliées

Que devient l'exception d'une opération que personne n'attend ? Le programme [Async12.java] le montre :

// une opération lancée puis oubliée : son exception est perdue
void main() {
    // le gestionnaire des exceptions qu'aucun code n'a interceptées dans un thread
    Thread.setDefaultUncaughtExceptionHandler((thread, e) ->
            IO.println("exception non interceptée dans [" + thread.getName() + "] : " + e.getMessage()));

    // 1 - un CompletableFuture attendu : son exception est interceptée
    try {
        enregistrer("a").join();
    } catch (CompletionException e) {
        IO.println("1 - CompletableFuture attendu : exception interceptée : " + e.getCause().getMessage());
    }

    // 2 - un CompletableFuture oublié : son exception reste dans l'objet, personne ne la verra
    enregistrer("b");
    IO.println("2 - CompletableFuture oublié : l'appel a rendu la main");

    // 3 - une tâche soumise à un exécuteur et oubliée : son exception reste dans le Future
    ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor();
    executeur.submit(() -> enregistrerBloquant("c"));
    IO.println("3 - Future oublié : la tâche est lancée");

    // 4 - un thread : l'exception non interceptée termine le thread et arrive au gestionnaire
    Thread.ofVirtual().name("enregistreur").start(() -> enregistrerBloquant("d"));
    IO.println("4 - thread virtuel lancé");

    // 5 - lancer sans attendre, mais en traitant l'échec
    enregistrer("e").exceptionally(e -> {
        IO.println("5 - échec signalé par exceptionally : " + e.getCause().getMessage());
        return null;
    });

    // le programme continue son travail…
    for (int i = 1; i <= 4; i++) {
        dormir(100);
        IO.println("le programme travaille (" + i + ")");
    }
    executeur.close();
    IO.println("fin normale du programme");
}

// une opération asynchrone qui échoue au bout de 200 ms
CompletableFuture<Void> enregistrer(String donnee) {
    return new CompletableFuture<Void>()
            .completeOnTimeout(null, 200, TimeUnit.MILLISECONDS)
            .thenRun(() -> {
                throw new IllegalStateException("échec de l'enregistrement de [" + donnee + "]");
            });
}

// la même opération, écrite de façon bloquante
void enregistrerBloquant(String donnee) {
    dormir(200);
    throw new IllegalStateException("échec de l'enregistrement de [" + donnee + "]");
}

// Thread.sleep sans exception contrôlée : une interruption devient une CancellationException
void dormir(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("attente interrompue");
    }
}

Résultats :

> java Async12.java
1 - CompletableFuture attendu : exception interceptée : échec de l'enregistrement de [a]
2 - CompletableFuture oublié : l'appel a rendu la main
3 - Future oublié : la tâche est lancée
4 - thread virtuel lancé
le programme travaille (1)
5 - échec signalé par exceptionally : échec de l'enregistrement de [e]
exception non interceptée dans [enregistreur] : échec de l'enregistrement de [d]
le programme travaille (2)
le programme travaille (3)
le programme travaille (4)
fin normale du programme

Commentaires :

• lignes 4-5 : [Thread.setDefaultUncaughtExceptionHandler] installe le gestionnaire appelé lorsqu'une exception n'est interceptée par aucun code d'un thread, juste avant la fin de ce thread. Sans lui, la JVM affiche la trace de pile de l'exception sur la sortie d'erreur ;
• lignes 8-12 : l'opération est attendue, son exception est interceptée normalement ;
• lignes 14-16 : l'opération est lancée puis oubliée. Son exception est rangée dans le [CompletableFuture], que plus personne ne consultera : elle est perdue, sans le moindre message ;
• lignes 18-21 : il en est de même pour une tâche soumise par [submit] à un exécuteur : son exception est rangée dans le [Future] rendu, qui est ignoré ;
• ligne 24 : un thread lancé directement n'a pas de [Future] : l'exception qu'il ne traite pas termine le thread et arrive au gestionnaire des exceptions non interceptées (ligne 8 des résultats). Le programme, lui, continue ;
• lignes 28-31 : la bonne façon de lancer une opération sans l'attendre : on lui attache une continuation [exceptionally] (ou [whenComplete]) qui traite l'échec, ici en l'affichant (ligne 7) ;
• les lignes 33-39 montrent que, dans tous les cas, le programme poursuit son travail jusqu'à sa fin normale. C'est une différence avec certains langages, où une exception perdue peut arrêter le programme ; en Java, elle passe inaperçue, ce qui n'est pas mieux : une erreur d'enregistrement que personne ne voit est une donnée perdue.

La règle est donc : une opération lancée doit être attendue ([join], [get], [allOf], ou la méthode [close] d'un exécuteur, qui attend ses tâches), ou recevoir une continuation qui traite son échec. Les outils d'analyse de code (SpotBugs, l'analyse d'IntelliJ IDEA) peuvent d'ailleurs signaler le résultat ignoré d'une méthode qui rend un [Future].

9.6. Annulation et délais

9.6.1. Future.cancel et l'interruption des threads

Une opération longue doit pouvoir être annulée : l'utilisateur a cliqué sur [Annuler], le client d'un serveur s'est déconnecté, l'application s'arrête, un délai est dépassé. Java utilise pour cela un mécanisme coopératif, l'interruption des threads :

• la méthode [interrupt()] d'un thread positionne son indicateur d'interruption ; elle n'arrête pas le thread (la méthode [Thread.stop], qui le faisait de force, est inutilisable depuis Java 20 : elle lance une [UnsupportedOperationException]) ;
• si le thread est bloqué dans une attente interruptible ([Thread.sleep], [Object.wait], [Thread.join], [BlockingQueue.take], [Future.get]…), l'attente s'interrompt aussitôt et lance une [InterruptedException], en effaçant l'indicateur ;
• un thread qui calcule consulte lui-même son indicateur, régulièrement, avec [Thread.currentThread().isInterrupted()], et s'arrête proprement s'il est positionné ;
• la méthode [cancel(true)] d'un [Future] rendu par un exécuteur fait passer la tâche à l'état [CANCELLED] et interrompt le thread qui l'exécute ([cancel(false)] se contente d'empêcher la tâche de démarrer si elle ne l'a pas encore fait). Après l'annulation, [get()] lance une [CancellationException].

Le mécanisme est coopératif : personne n'arrête de force une opération ; l'opération s'arrête elle-même, à un endroit où elle peut le faire proprement. C'est pourquoi une méthode qui intercepte une [InterruptedException] ne doit pas l'ignorer : elle doit soit la propager, soit s'arrêter, soit au minimum rétablir l'indicateur avec [Thread.currentThread().interrupt()] pour que le code appelant sache qu'une interruption a été demandée (c'est ce que fait notre méthode [dormir]).

Le programme [Async13.java] annule de plusieurs façons un téléchargement simulé :

// annuler une opération : Future.cancel(true) et l'interruption des threads
long debut = System.nanoTime();

void main() throws InterruptedException, ExecutionException {
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        // 1 - on n'attend pas le résultat plus de 1,2 seconde, puis on annule l'opération
        Future<?> telechargement1 = executeur.submit(() -> telecharger(10));
        try {
            telechargement1.get(1200, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            // cancel(true) interrompt le thread qui exécute la tâche
            telechargement1.cancel(true);
            afficher("1 - " + e.getClass().getSimpleName() + ", état : " + telechargement1.state());
        }

        // 2 - un calcul qui vérifie lui-même, régulièrement, s'il a été interrompu
        debut = System.nanoTime();
        Future<Long> calcul = executeur.submit(this::calculer);
        Thread.sleep(300);
        calcul.cancel(true);
        try {
            afficher("2 - résultat : " + calcul.get());
        } catch (CancellationException e) {
            afficher("2 - get() lance " + e.getClass().getSimpleName() + ", état : " + calcul.state());
        }
        Thread.sleep(100);

        // 3 - l'annulation est demandée par le programme au bout de 700 ms
        debut = System.nanoTime();
        Future<?> telechargement3 = executeur.submit(() -> telecharger(10));
        Thread.sleep(700);
        afficher("3 - le programme demande l'annulation");
        telechargement3.cancel(true);
        afficher("3 - isCancelled=" + telechargement3.isCancelled());
        Thread.sleep(100);

        // 4 - cancel() d'un CompletableFuture n'interrompt pas l'opération qu'il représente
        debut = System.nanoTime();
        CompletableFuture<Void> telechargement4 = CompletableFuture.runAsync(() -> telecharger(3), executeur);
        Thread.sleep(400);
        telechargement4.cancel(true);
        afficher("4 - CompletableFuture annulé : isCancelled=" + telechargement4.isCancelled()
                + ", mais l'opération continue");
    } // close() attend la fin de toutes les tâches
}

// simule le téléchargement de [nbBlocs] blocs de données (un bloc toutes les 300 ms)
void telecharger(int nbBlocs) {
    try {
        for (int i = 1; i <= nbBlocs; i++) {
            // une attente bloquante est interrompue dès que le thread est interrompu
            Thread.sleep(300);
            afficher("  bloc " + i + "/" + nbBlocs + " reçu");
        }
    } catch (InterruptedException e) {
        afficher("  téléchargement interrompu");
    }
}

// un long calcul qui consulte l'indicateur d'interruption de son thread tous les 1 000 000 tours
long calculer() {
    long somme = 0;
    for (long i = 0; i < 100_000_000_000L; i++) {
        if (i % 1_000_000 == 0 && Thread.currentThread().isInterrupted()) {
            afficher("  calcul interrompu au tour " + i);
            return -1;
        }
        somme += i % 7;
    }
    return somme;
}

// affichage avec le temps écoulé depuis [debut]
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async13.java
  318 ms :   bloc 1/10 reçu
  638 ms :   bloc 2/10 reçu
  939 ms :   bloc 3/10 reçu
 1212 ms :   téléchargement interrompu
 1214 ms : 1 - TimeoutException, état : CANCELLED
  301 ms : 2 - get() lance CancellationException, état : CANCELLED
  303 ms :   calcul interrompu au tour 131000000
  302 ms :   bloc 1/10 reçu
  603 ms :   bloc 2/10 reçu
  701 ms : 3 - le programme demande l'annulation
  702 ms :   téléchargement interrompu
  702 ms : 3 - isCancelled=true
  309 ms :   bloc 1/3 reçu
  404 ms : 4 - CompletableFuture annulé : isCancelled=true, mais l'opération continue
  609 ms :   bloc 2/3 reçu
  910 ms :   bloc 3/3 reçu

Commentaires :

• lignes 47-58 : le téléchargement simulé reçoit un bloc toutes les 300 ms. Il est exécuté par un thread virtuel et utilise une attente bloquante, [Thread.sleep], qui s'interrompt dès que le thread est interrompu : l'[InterruptedException] fait sortir de la boucle, et le téléchargement s'arrête en le signalant. Il n'a besoin d'aucun paramètre particulier pour être annulable : les attentes de [java.util.concurrent] réagissent de même à l'interruption, ainsi que, dans un thread virtuel, les E/S bloquantes des sockets (le socket est alors fermé et l'opération lance une [SocketException]) ;
• lignes 6-14 : [get(1200, TimeUnit.MILLISECONDS)] attend le résultat au plus 1,2 seconde, puis lance une [TimeoutException]. L'attente est abandonnée mais pas l'opération : on l'annule explicitement avec [cancel(true)], qui interrompt le thread virtuel du téléchargement. Celui-ci a reçu trois blocs et s'arrête à 1 200 ms (lignes 2-6 des résultats) ;
• lignes 16-26 : un calcul (et non une attente) annulable. La méthode [calculer] (lignes 60-71) consulte l'indicateur d'interruption de son thread tous les millions de tours et s'arrête s'il est positionné. Le programme l'annule au bout de 300 ms : [get()] lance aussitôt une [CancellationException] (ligne 7), et le calcul s'arrête de lui-même peu après (ligne 8) ;
• lignes 28-35 : l'annulation est demandée par le programme au bout de 700 ms. Elle est immédiate : l'attente du troisième bloc est interrompue (lignes 11-13) ;
• lignes 37-43 : un piège à connaître. La méthode [cancel] d'un [CompletableFuture] termine celui-ci dans l'état [CANCELLED], mais n'interrompt rien : un [CompletableFuture] ne sait pas quel thread, s'il y en a un, travaille pour lui (son paramètre [mayInterruptIfRunning] est ignoré). Le téléchargement continue jusqu'à son terme (lignes 16-17) ; seul son résultat sera ignoré. Pour qu'un [CompletableFuture] puisse arrêter l'opération qu'il représente, il faut que celle-ci surveille l'état du [CompletableFuture] ([isCancelled()]), ou que l'on garde le [Future] ou le thread de la tâche pour l'interrompre ;
• ligne 44 : la méthode [close()] de l'exécuteur attend la fin de toutes les tâches, y compris celle de l'étape 4.

9.6.2. Plusieurs traitements, plusieurs issues

Le programme [Async14.java] lance cinq traitements de 1 à 5 secondes : le n° 3 lance une exception, le n° 4 est annulé. Il exploite chaque traitement dès qu'il est terminé, quelle que soit la façon dont il s'est terminé, et sait toujours de quel traitement il s'agit :

// cinq traitements de 1 à 5 secondes exécutés par des threads virtuels :
// le n° 3 lance une exception, le n° 4 est annulé
long debut;

void main() throws InterruptedException {
    debut = System.nanoTime();
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor();
         ScheduledExecutorService planificateur = Executors.newSingleThreadScheduledExecutor()) {
        // le CompletionService livrera chaque tâche dès qu'elle sera terminée
        CompletionService<Resultat> service = new ExecutorCompletionService<>(executeur);
        List<Future<Resultat>> taches = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            int numero = i;
            taches.add(service.submit(() -> dormir(numero, numero + 1)));
        }
        // on annule le traitement n° 4 au bout de 2,5 secondes
        planificateur.schedule(() -> taches.get(4).cancel(true), 2500, TimeUnit.MILLISECONDS);
        afficher("les " + taches.size() + " traitements sont lancés");

        // on exploite chaque traitement dès qu'il est terminé, quelle que soit sa façon de se terminer
        for (int i = 0; i < taches.size(); i++) {
            Future<Resultat> tache = service.take();
            int numero = taches.indexOf(tache);
            String bilan = switch (tache.state()) {
                case SUCCESS -> String.format("terminé : durée programmée %d s, durée effective %.1f s",
                        tache.resultNow().duree(), tache.resultNow().dureeEffective());
                case FAILED -> "a rencontré l'erreur suivante : " + tache.exceptionNow().getMessage();
                case CANCELLED -> "annulé";
                case RUNNING -> "toujours en cours ?";
            };
            afficher("traitement " + numero + " " + bilan);
        }
    }
}

// le traitement : dormir [duree] secondes, par tranches d'une seconde
Resultat dormir(int numero, int duree) throws InterruptedException {
    long debutTraitement = System.nanoTime();
    // exception pour le traitement n° 3
    if (numero == 3) {
        throw new IllegalStateException("test....");
    }
    for (int i = 1; i <= duree; i++) {
        Thread.sleep(1000);
    }
    // le résultat rendu à l'appelant
    return new Resultat(duree, (System.nanoTime() - debutTraitement) / 1e9);
}

// le résultat d'un traitement
record Resultat(int duree, double dureeEffective) {
}

// affichage avec le temps écoulé depuis le début, en secondes
void afficher(String message) {
    IO.println(String.format("%4.1f s : %s", (System.nanoTime() - debut) / 1e9, message));
}

Résultats :

> java Async14.java
 0,0 s : les 5 traitements sont lancés
 0,0 s : traitement 3 a rencontré l'erreur suivante : test....
 1,0 s : traitement 0 terminé : durée programmée 1 s, durée effective 1,0 s
 2,0 s : traitement 1 terminé : durée programmée 2 s, durée effective 2,0 s
 2,5 s : traitement 4 annulé
 3,0 s : traitement 2 terminé : durée programmée 3 s, durée effective 3,0 s

Commentaires :

• lignes 36-48 : le traitement. Le n° 3 lance une exception ; les autres « dorment » [duree] secondes par tranches d'une seconde, avec un [Thread.sleep] interruptible, et rendent un résultat, un [record] qui contient la durée programmée et la durée effective. La méthode déclare [InterruptedException] : elle est appelée par un [Callable], dont la méthode [call] peut lancer des exceptions contrôlées ;
• lignes 7-8 : deux exécuteurs déclarés dans le même [try] : l'exécuteur des threads virtuels, et un [ScheduledExecutorService], un exécuteur capable d'exécuter une tâche après un délai (nous le retrouverons au paragraphe sur le temps) ;
• lignes 10-15 : les cinq traitements sont soumis au [CompletionService] déjà rencontré dans [Async10.java], et leurs [Future] sont rangés dans une liste ;
• ligne 17 : le [ScheduledExecutorService] annulera le traitement n° 4 au bout de 2,5 secondes, alors qu'il en a encore 2,5 à dormir ;
• lignes 21-32 : [take()] livre chaque traitement dès qu'il est terminé, quelle que soit la façon dont il s'est terminé : un [Future] annulé est livré lui aussi. On n'appelle pas [get()], qui lancerait une exception, mais on examine l'état du [Future] avec une expression [switch] qui couvre les quatre constantes de [Future.State] (le compilateur vérifie qu'aucune ne manque). Le numéro du traitement est retrouvé par la position de son [Future] dans la liste ([indexOf]) ;
• les résultats montrent que le n° 3 échoue tout de suite, que les n° 0, 1 et 2 se terminent au bout de 1, 2 et 3 secondes, et que le n° 4 est annulé à 2,5 secondes.

9.6.3. Limiter la durée d'une attente

Le programme [Async15.java] présente les outils qui limitent la durée d'une attente : les méthodes [orTimeout] et [completeOnTimeout] de [CompletableFuture] (Java 9), et la méthode [get] avec délai d'un [Future]. Il montre aussi comment combiner deux causes d'arrêt.

// limiter la durée d'une attente : orTimeout, completeOnTimeout, get(délai)
long debut = System.nanoTime();

void main() throws InterruptedException, ExecutionException {
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        // 1 - orTimeout : on n'attend pas plus d'une seconde
        CompletableFuture<String> lente = CompletableFuture.supplyAsync(
                () -> interrogerBloquant("serveur-lent", 3000), executeur);
        try {
            afficher("1 - réponse : " + lente.orTimeout(1, TimeUnit.SECONDS).join());
        } catch (CompletionException e) {
            afficher("1 - " + e.getCause().getClass().getSimpleName() + ", état : " + lente.state());
        }
        // l'opération, elle, continue : on la laisse finir
        Thread.sleep(2200);

        // 2 - completeOnTimeout : au-delà d'une seconde, on se contente d'une réponse par défaut
        debut = System.nanoTime();
        String reponse = interroger("serveur-lent", 3000)
                .completeOnTimeout("réponse par défaut", 1, TimeUnit.SECONDS)
                .join();
        afficher("2 - " + reponse);

        // 3 - pour arrêter vraiment l'opération, on interrompt le thread qui l'exécute
        debut = System.nanoTime();
        Future<String> lente3 = executeur.submit(() -> interrogerBloquant("serveur-lent", 3000));
        try {
            afficher("3 - réponse : " + lente3.get(1, TimeUnit.SECONDS));
        } catch (TimeoutException e) {
            lente3.cancel(true);
            afficher("3 - délai d'une seconde dépassé, opération annulée, état : " + lente3.state());
        }
    }

    // 4 - deux causes d'arrêt : le délai propre à chaque traitement et un arrêt général
    debut = System.nanoTime();
    CompletableFuture<Void> arretGeneral = new CompletableFuture<>();
    CompletableFuture<Void> tous = CompletableFuture.allOf(
            traiterAvecDelai("rapide", 500, 1000, arretGeneral),
            traiterAvecDelai("lent", 3000, 1000, arretGeneral),
            traiterAvecDelai("moyen", 1500, 2000, arretGeneral));
    // l'appelant demande l'arrêt général au bout de 1 200 ms
    Thread.sleep(1200);
    afficher("4 - arrêt général demandé");
    arretGeneral.complete(null);
    tous.join();
}

// un traitement limité à [delai] millisecondes, qui peut aussi être arrêté par l'appelant
CompletableFuture<Void> traiterAvecDelai(String nom, int duree, int delai, CompletableFuture<Void> arret) {
    CompletableFuture<String> reponse = interroger(nom, duree).orTimeout(delai, TimeUnit.MILLISECONDS);
    // l'arrêt général annule l'attente de la réponse
    arret.thenRun(() -> reponse.cancel(false));
    // on examine la façon dont l'attente s'est terminée
    return reponse.handle((texte, e) -> switch (e) {
        case null -> texte;
        case TimeoutException _ -> nom + " : délai de " + delai + " ms dépassé";
        case CancellationException _ -> nom + " : arrêté à la demande de l'appelant";
        default -> nom + " : " + e;
    }).thenAccept(message -> afficher("4 - " + message));
}

// simule un serveur qui répond au bout de [duree] millisecondes
CompletableFuture<String> interroger(String serveur, int duree) {
    return new CompletableFuture<String>()
            .completeOnTimeout(serveur + " a répondu en " + duree + " ms", duree, TimeUnit.MILLISECONDS);
}

// la même opération, écrite de façon bloquante
String interrogerBloquant(String serveur, int duree) {
    try {
        Thread.sleep(duree);
        afficher("  " + serveur + " a fini son travail");
        return serveur + " a répondu en " + duree + " ms";
    } catch (InterruptedException e) {
        afficher("  " + serveur + " a été interrompu");
        throw new CancellationException("opération interrompue");
    }
}

// affichage avec le temps écoulé depuis [debut]
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async15.java
 1028 ms : 1 - TimeoutException, état : FAILED
 3025 ms :   serveur-lent a fini son travail
 1001 ms : 2 - réponse par défaut
 1003 ms :   serveur-lent a été interrompu
 1004 ms : 3 - délai d'une seconde dépassé, opération annulée, état : CANCELLED
  501 ms : 4 - rapide a répondu en 500 ms
 1005 ms : 4 - lent : délai de 1000 ms dépassé
 1206 ms : 4 - arrêt général demandé
 1207 ms : 4 - moyen : arrêté à la demande de l'appelant

Commentaires :

• lignes 7-8 : l'opération lente est une méthode bloquante (lignes 69-79), exécutée par un thread virtuel grâce à [supplyAsync] avec un exécuteur ;
• ligne 10 : [orTimeout(délai, unité)] termine le [CompletableFuture] avec une [TimeoutException] s'il n'est pas terminé dans le délai. C'est l'écriture directe de la « course » avec [anyOf] du programme [Async09.java]. [orTimeout] ne crée pas un nouveau [CompletableFuture] : elle modifie celui auquel elle s'applique (et le rend) ; c'est pourquoi l'état de [lente] est [FAILED] (ligne 2 des résultats) ;
• lignes 14-15 : [orTimeout] abandonne l'attente, pas l'opération : le thread virtuel continue de travailler et termine son travail à 3 secondes (ligne 3), pour rien. C'est le défaut déjà vu avec [cancel] : un [CompletableFuture] ne peut pas interrompre l'opération qu'il représente ;
• lignes 17-22 : [completeOnTimeout(valeur, délai, unité)] termine le [CompletableFuture] avec une valeur par défaut s'il n'est pas terminé dans le délai : au bout d'une seconde, on se contente d'une « réponse par défaut » (ligne 4) ;
• lignes 24-32 : pour arrêter vraiment l'opération, on la soumet à l'exécuteur, on attend son [Future] au plus une seconde, puis on l'annule avec [cancel(true)], qui interrompt son thread : l'opération s'arrête à 1 seconde (lignes 5-6). C'est la solution à préférer lorsque l'opération est bloquante et interruptible ;
• lignes 49-61 : un traitement peut avoir plusieurs raisons d'être arrêté, ici son propre délai et une demande d'arrêt général de l'appelant :
  • ligne 51 : le délai propre au traitement est posé avec [orTimeout] ;
  • ligne 53 : l'arrêt général est lui-même un [CompletableFuture<Void>], que l'appelant terminera ; on lui attache une continuation qui annule l'attente de la réponse ;
  • lignes 55-60 : quelle que soit la cause, l'attente se termine en échec ; [handle] examine l'exception avec un [switch] sur les types (Java 21) : [case null] pour le succès, [case TimeoutException _] pour le délai, [case CancellationException _] pour l'arrêt général. Le caractère [_] (Java 22) est une variable anonyme : on teste le type sans avoir besoin de nommer l'objet ;
• lignes 36-46 : trois traitements et un arrêt général demandé à 1 200 ms en terminant le [CompletableFuture] [arretGeneral]. Le traitement [rapide] se termine normalement, le traitement [lent] dépasse son délai d'une seconde, le traitement [moyen], qui avait droit à 2 secondes, est arrêté par l'arrêt général (lignes 7-10).

9.7. Signaler la progression

Une opération longue doit souvent signaler son avancement (une barre de progression, un pourcentage). Le plus simple est de lui passer une fonction de rappel qui reçoit les rapports de progression : un [Consumer<T>], l'interface fonctionnelle du chapitre 7 dont l'unique méthode [accept(T)] consomme une valeur. L'opération appelle [accept] à chaque étape ; elle ignore tout de la façon dont les rapports seront affichés.

Dans une interface graphique, les rapports, émis par le thread qui exécute l'opération, ne peuvent pas mettre à jour directement les composants graphiques : le [Consumer] doit confier la mise à jour au thread de l'interface ([SwingUtilities.invokeLater(() -> barre.setValue(pourcentage))] avec Swing, [Platform.runLater] avec JavaFX). Dans un programme console, il peut afficher directement. Le programme [Async16.java] simule une copie de fichiers que l'utilisateur peut annuler en tapant [Entrée] :

// signaler la progression d'une opération avec un Consumer ; l'utilisateur peut l'annuler
void main() throws InterruptedException {
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        IO.println("Copie en cours, tapez [Entrée] pour l'annuler");
        // la copie est exécutée par un thread virtuel ; la méthode afficher affiche ses rapports
        Future<Integer> copie = executeur.submit(() -> copier(8, this::afficher));
        // un autre thread virtuel attend que l'utilisateur tape [Entrée] pour annuler la copie
        Thread.ofVirtual().start(() -> {
            // readln rend null si l'entrée standard est fermée : on n'annule rien
            if (IO.readln() != null) {
                copie.cancel(true);
            }
        });
        try {
            IO.println("Copie terminée : " + copie.get() + " fichiers copiés");
        } catch (CancellationException e) {
            IO.println("Copie annulée par l'utilisateur");
        } catch (ExecutionException e) {
            IO.println("Échec de la copie : " + e.getCause());
        }
    }
}

// l'opération ne sait pas comment sa progression est affichée : elle la signale seulement
int copier(int nbFichiers, Consumer<Avancement> progression) throws InterruptedException {
    for (int i = 1; i <= nbFichiers; i++) {
        // la copie d'un fichier
        Thread.sleep(300);
        // le rapport de progression (l'appelant peut ne pas en vouloir : progression vaut alors null)
        if (progression != null) {
            progression.accept(new Avancement(100 * i / nbFichiers, "fichier" + i + ".dat"));
        }
    }
    return nbFichiers;
}

// un rapport de progression
record Avancement(int pourcentage, String fichier) {
}

// l'affichage des rapports de progression dans la console : une barre de 20 caractères
void afficher(Avancement avancement) {
    int nbDieses = avancement.pourcentage() / 5;
    String barre = "#".repeat(nbDieses) + ".".repeat(20 - nbDieses);
    IO.println(String.format("[%s] %3d %% %s", barre, avancement.pourcentage(), avancement.fichier()));
}

Résultats sans annulation :

> java Async16.java
Copie en cours, tapez [Entrée] pour l'annuler
[##..................]  12 % fichier1.dat
[#####...............]  25 % fichier2.dat
[#######.............]  37 % fichier3.dat
[##########..........]  50 % fichier4.dat
[############........]  62 % fichier5.dat
[###############.....]  75 % fichier6.dat
[#################...]  87 % fichier7.dat
[####################] 100 % fichier8.dat
Copie terminée : 8 fichiers copiés

Résultats lorsque l'utilisateur tape [Entrée] pendant la copie, au bout d'une seconde et demie environ :

> java Async16.java
Copie en cours, tapez [Entrée] pour l'annuler
[##..................]  12 % fichier1.dat
[#####...............]  25 % fichier2.dat
[#######.............]  37 % fichier3.dat
[##########..........]  50 % fichier4.dat
[############........]  62 % fichier5.dat
Copie annulée par l'utilisateur

Commentaires :

• lignes 24-35 : l'opération reçoit un [Consumer<Avancement>]. Après chaque fichier, elle signale son avancement par un objet [Avancement] (un [record] avec le pourcentage et le nom du fichier). Un appelant qui ne veut pas de rapports passe [null], d'où le test de la ligne 30 ;
• lignes 41-46 : l'affichage d'un rapport : une barre de 20 caractères construite avec [String.repeat] (Java 11). La référence de méthode [this::afficher] (ligne 6) est le [Consumer<Avancement>] passé à l'opération ;
• ligne 6 : la copie est exécutée par un thread virtuel ;
• lignes 8-13 : un autre thread virtuel attend que l'utilisateur tape [Entrée] ([IO.readln()] est une lecture bloquante, d'où ce thread : c'est lui qui est bloqué, pas le programme), puis annule la copie avec [cancel(true)], qui interrompt le [Thread.sleep] de la ligne 28. Le programme ne l'attend pas : c'est un thread démon, abandonné à la fin du programme si l'utilisateur n'a rien tapé ;
• lignes 14-20 : le programme attend la fin de la copie. Une annulation se traduit par une [CancellationException], un échec de la copie par une [ExecutionException].

9.8. Les entrées / sorties asynchrones

9.8.1. Les fichiers

Les E/S sont la raison d'être de l'asynchronisme. Java propose pour les fichiers deux familles de classes :

• les classes « classiques », bloquantes : [Files] ([readAllLines], [readString], [lines], [writeString], [newBufferedReader], [newBufferedWriter]…), [BufferedReader], [BufferedWriter]. On les exécute aujourd'hui sur un thread virtuel lorsqu'on ne veut pas bloquer un thread important ;
• la classe [AsynchronousFileChannel] (Java 7), dont les opérations rendent un [Future] ou appellent un [CompletionHandler]. Elle est peu utilisée : sous Linux, elle exécute en fait des lectures bloquantes sur un pool de threads interne à la JVM, et son interface est malcommode.

Le programme [Async17.java] présente les deux familles en lisant le fichier des tranches d'impôt du fil rouge, [DataImpot.txt], placé dans le dossier du programme :

4962:0:0
8382:0.068:291.09
14753:0.191:1322.92
23888:0.283:2668.39
38868:0.374:4846.98
47932:0.426:6883.66
0:0.481:9505.54

Chaque ligne contient la limite de la tranche, le coefficient [coeffR] et le coefficient [coeffN], séparés par le caractère [:] et écrits avec un point décimal (le barème est celui de 2004 sur les revenus de 2003, conservé tel quel pour l'exercice). Le programme est exécuté dans le dossier [09-async] et cherche ses fichiers dans le dossier courant.

// les entrées / sorties sur fichiers : AsynchronousFileChannel, puis des E/S bloquantes sur un thread virtuel
final Path donnees = Path.of("DataImpot.txt");
final Path rapport = Path.of("rapport-17.txt");

void main() throws InterruptedException {
    // 1 - AsynchronousFileChannel (Java 7) : une lecture asynchrone, rendue comme un CompletableFuture
    String contenu = lireAsynchrone(donnees).join();
    IO.println("1 - AsynchronousFileChannel : " + contenu.lines().count() + " lignes");

    // 2 à 7 - des E/S bloquantes ordinaires, exécutées par un thread virtuel
    Thread.ofVirtual().start(() -> {
        try {
            traiterFichiers();
        } catch (IOException e) {
            IO.println("erreur : " + e);
        }
    }).join();
}

// lit tout un fichier (de petite taille) avec un AsynchronousFileChannel
CompletableFuture<String> lireAsynchrone(Path fichier) {
    CompletableFuture<String> resultat = new CompletableFuture<>();
    try {
        AsynchronousFileChannel canal = AsynchronousFileChannel.open(fichier, StandardOpenOption.READ);
        ByteBuffer tampon = ByteBuffer.allocate((int) canal.size());
        // la lecture est lancée ; l'objet de rappel terminera le CompletableFuture
        canal.read(tampon, 0, tampon, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer nbOctets, ByteBuffer lus) {
                resultat.complete(new String(lus.array(), 0, nbOctets, StandardCharsets.UTF_8));
            }

            @Override
            public void failed(Throwable e, ByteBuffer lus) {
                resultat.completeExceptionally(e);
            }
        });
        // le canal sera fermé quand la lecture sera terminée, quelle qu'en soit l'issue
        resultat.whenComplete((texte, e) -> fermer(canal));
    } catch (IOException e) {
        resultat.completeExceptionally(e);
    }
    return resultat;
}

void traiterFichiers() throws IOException {
    // 2 - lecture de tout le fichier en une seule opération
    List<String> lignes = Files.readAllLines(donnees);
    IO.println("2 - Files.readAllLines : " + lignes.size() + " lignes");

    // 3 - lecture ligne par ligne avec un BufferedReader
    List<TrancheImpot> tranches = new ArrayList<>();
    try (BufferedReader lecteur = Files.newBufferedReader(donnees)) {
        String ligne;
        // readLine rend null à la fin du fichier
        while ((ligne = lecteur.readLine()) != null) {
            tranches.add(TrancheImpot.parse(ligne));
        }
    }
    IO.println("3 - BufferedReader.readLine : " + tranches.size() + " tranches, 2e : " + tranches.get(1));

    // 4 - Files.lines rend un Stream de lignes, lues à la demande
    try (Stream<String> flux = Files.lines(donnees)) {
        IO.println("4 - Files.lines : " + flux.count() + " lignes");
    }

    // 5 - écriture avec un BufferedWriter
    try (BufferedWriter ecrivain = Files.newBufferedWriter(rapport)) {
        ecrivain.write("Tranches du barème (exercice : barème 2004 sur les revenus 2003)");
        ecrivain.newLine();
        for (int i = 0; i < tranches.size(); i++) {
            TrancheImpot tranche = tranches.get(i);
            String limite = tranche.limite().signum() == 0 ? "au-delà"
                    : String.format("jusqu'à %,d €", tranche.limite().intValue());
            ecrivain.write(String.format("tranche %d : %-18s taux %.1f %%",
                    i + 1, limite, tranche.coeffR().multiply(BigDecimal.valueOf(100))));
            ecrivain.newLine();
        }
    }

    // 6 - lecture de tout le fichier sous la forme d'une chaîne
    String texte = Files.readString(rapport);
    IO.print("5 et 6 - contenu de [" + rapport + "] :\n" + texte);

    // 7 - ajout en fin de fichier puis suppression
    Files.writeString(rapport, "fin du rapport\n", StandardOpenOption.APPEND);
    IO.println("7 - dernière ligne : " + Files.readAllLines(rapport).getLast());
    Files.delete(rapport);
}

// ferme un canal sans exception contrôlée
void fermer(Channel canal) {
    try {
        canal.close();
    } catch (IOException e) {
        IO.println("erreur à la fermeture : " + e);
    }
}

// une tranche d'impôt ; le fichier utilise le point décimal, que BigDecimal comprend
record TrancheImpot(BigDecimal limite, BigDecimal coeffR, BigDecimal coeffN) {
    static TrancheImpot parse(String ligne) {
        String[] champs = ligne.split(":");
        return new TrancheImpot(new BigDecimal(champs[0]), new BigDecimal(champs[1]),
                new BigDecimal(champs[2]));
    }
}

Résultats :

> java Async17.java
1 - AsynchronousFileChannel : 7 lignes
2 - Files.readAllLines : 7 lignes
3 - BufferedReader.readLine : 7 tranches, 2e : TrancheImpot[limite=8382, coeffR=0.068, coeffN=291.09]
4 - Files.lines : 7 lignes
5 et 6 - contenu de [rapport-17.txt] :
Tranches du barème (exercice : barème 2004 sur les revenus 2003)
tranche 1 : jusqu'à 4 962 €    taux 0,0 %
tranche 2 : jusqu'à 8 382 €    taux 6,8 %
tranche 3 : jusqu'à 14 753 €   taux 19,1 %
tranche 4 : jusqu'à 23 888 €   taux 28,3 %
tranche 5 : jusqu'à 38 868 €   taux 37,4 %
tranche 6 : jusqu'à 47 932 €   taux 42,6 %
tranche 7 : au-delà            taux 48,1 %
7 - dernière ligne : fin du rapport

Commentaires :

• lignes 20-44 : la lecture avec un [AsynchronousFileChannel], présentée sous la forme d'un [CompletableFuture<String>]. C'est la technique générale pour transformer une API à méthodes de rappel en [CompletableFuture] : on crée un [CompletableFuture] « en cours » (ligne 22), on le rend aussitôt (ligne 43), et ce sont les méthodes de rappel qui le terminent, avec le résultat ([complete], ligne 30) ou l'exception ([completeExceptionally], ligne 35). L'objet de rappel est une classe anonyme qui implémente [CompletionHandler<Integer, ByteBuffer>] : le résultat d'une lecture est le nombre d'octets lus, et l'objet transmis est ici le tampon lui-même. Une seule lecture suffit pour un petit fichier ; un gros fichier demanderait des lectures successives, que les méthodes de rappel devraient enchaîner ;
• ligne 39 : [whenComplete] ferme le canal quand la lecture est terminée, qu'elle ait réussi ou non ;
• lignes 11-17 : les autres opérations sont des E/S bloquantes ordinaires, regroupées dans la méthode [traiterFichiers] et exécutées par un thread virtuel. La lambda d'un [Runnable] ne peut pas lancer d'exception contrôlée : l'[IOException] est interceptée dans la lambda ;
• ligne 48 : [Files.readAllLines] lit tout le fichier et rend la liste de ses lignes (en UTF-8 par défaut, comme toutes les méthodes de [Files]) ;
• lignes 52-59 : la lecture ligne par ligne avec un [BufferedReader] ; [readLine()] rend [null] à la fin du fichier. Chaque ligne est transformée en [TrancheImpot] par la méthode [parse] du record (lignes 100-107) ; les nombres sont des [BigDecimal], dont le constructeur attend un point décimal quelle que soit la langue de la machine. La représentation d'un record produite par sa méthode [toString] apparaît à la ligne 4 des résultats ;
• lignes 63-65 : [Files.lines] rend un [Stream<String>] de lignes, lues à la demande : contrairement à [readAllLines], il ne charge pas tout le fichier en mémoire. C'est la forme à préférer pour les gros fichiers. Le flux tient le fichier ouvert : il doit être fermé, d'où le [try] avec ressource ;
• lignes 68-79 : l'écriture avec un [BufferedWriter], fermé (et donc vidé sur le disque) à la sortie du [try]. Le format [%,d] insère le séparateur des milliers de la langue de la machine (une espace insécable fine en français) ;
• lignes 82-88 : [Files.readString] (Java 11) relit le fichier écrit, [Files.writeString] avec l'option [StandardOpenOption.APPEND] lui ajoute une ligne, [getLast()] (Java 21) rend le dernier élément de la liste, puis le fichier est supprimé.

Rappelons la limite indiquée plus haut : les lectures et écritures de fichiers épinglent le thread virtuel sur son thread porteur pendant l'opération, et la JVM compense en ajoutant un thread porteur. Pour les fichiers locaux, c'est rarement sensible ; pour le réseau, les threads virtuels libèrent réellement leur porteur pendant les attentes.

9.8.2. Le réseau : HttpClient

Le client HTTP de Java, [java.net.http.HttpClient] (Java 11), offre les deux façons de faire : [send], bloquante, qui rend la réponse, et [sendAsync], qui rend aussitôt un [CompletableFuture] de la réponse. Le chapitre 12 l'étudiera en détail ; le programme [Async18.java] se contente de comparer les deux méthodes en interrogeant un petit serveur HTTP local qui met une seconde à donner la « température » d'une ville :

// HttpClient : des requêtes HTTP asynchrones (sendAsync) ou bloquantes (send) vers un petit serveur local
import com.sun.net.httpserver.HttpServer;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

final int port = 27510;
final List<String> villes = List.of("Angers", "Brest", "Colmar", "Dijon", "Évry", "Foix");
long debut;

void main() throws IOException, InterruptedException, ExecutionException {
    // un serveur HTTP qui met une seconde à donner la « température » d'une ville
    HttpServer serveur = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), 0);
    serveur.createContext("/temperature", echange -> {
        // la requête est de la forme /temperature?ville=xxx
        String parametre = echange.getRequestURI().getRawQuery().substring("ville=".length());
        String ville = URLDecoder.decode(parametre, StandardCharsets.UTF_8);
        dormir(1000);
        byte[] reponse = (ville + " : " + (10 + ville.length()) + " °C").getBytes(StandardCharsets.UTF_8);
        echange.sendResponseHeaders(200, reponse.length);
        try (OutputStream flux = echange.getResponseBody()) {
            flux.write(reponse);
        }
    });
    // chaque requête reçue est traitée par un thread virtuel
    serveur.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
    serveur.start();

    try (HttpClient client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build()) {
        // une première requête, pour que la JVM charge les classes du client avant les mesures
        client.send(requete("Paris"), HttpResponse.BodyHandlers.discarding());

        // 1 - sendAsync rend aussitôt un CompletableFuture<HttpResponse<String>>
        debut = System.nanoTime();
        List<CompletableFuture<String>> reponses = villes.stream()
                .map(ville -> client.sendAsync(requete(ville), HttpResponse.BodyHandlers.ofString())
                        .thenApply(HttpResponse::body))
                .toList();
        CompletableFuture.allOf(reponses.toArray(CompletableFuture[]::new)).join();
        afficher("1 - sendAsync : " + reponses.stream().map(CompletableFuture::resultNow).toList());

        // 2 - send est bloquant : on l'appelle depuis des threads virtuels
        debut = System.nanoTime();
        try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> reponses2 = new ArrayList<>();
            for (String ville : villes) {
                reponses2.add(executeur.submit(
                        () -> client.send(requete(ville), HttpResponse.BodyHandlers.ofString()).body()));
            }
            List<String> textes = new ArrayList<>();
            for (Future<String> reponse : reponses2) {
                textes.add(reponse.get());
            }
            afficher("2 - send sur des threads virtuels : " + textes);
        }
    } finally {
        serveur.stop(0);
    }
}

// la requête GET qui demande la température d'une ville
HttpRequest requete(String ville) {
    String parametre = URLEncoder.encode(ville, StandardCharsets.UTF_8);
    URI adresse = URI.create("http://localhost:" + port + "/temperature?ville=" + parametre);
    return HttpRequest.newBuilder(adresse).build();
}

// Thread.sleep sans exception contrôlée : une interruption devient une CancellationException
void dormir(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("attente interrompue");
    }
}

// affichage avec le temps écoulé depuis [debut]
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async18.java
 1033 ms : 1 - sendAsync : [Angers : 16 °C, Brest : 15 °C, Colmar : 16 °C, Dijon : 15 °C, Évry : 14 °C, Foix : 14 °C]
 1023 ms : 2 - send sur des threads virtuels : [Angers : 16 °C, Brest : 15 °C, Colmar : 16 °C, Dijon : 15 °C, Évry : 14 °C, Foix : 14 °C]

Commentaires :

• lignes 12-27 : le serveur. La JVM contient un petit serveur HTTP, la classe [HttpServer] du module [jdk.httpserver], importée ligne 2 (un fichier compact n'importe automatiquement que le module [java.base] ; les classes du client, du module [java.net.http], sont importées lignes 3-5). [createContext] associe au chemin [/temperature] un gestionnaire, une lambda qui reçoit l'échange HTTP : elle décode le paramètre [ville] de l'adresse demandée, attend une seconde, puis envoie la réponse. La ligne 26 fait traiter chaque requête par un thread virtuel : un serveur ainsi configuré peut servir des milliers de clients qui attendent ;
• lignes 29-31 : le client est créé par un constructeur ([newBuilder]) qui fixe ici la version du protocole. Depuis Java 21, il implémente [AutoCloseable] et se déclare dans un [try] avec ressource. Une première requête sert à « chauffer » la JVM : la première utilisation d'un client charge de nombreuses classes, ce qui fausserait la mesure suivante de quelques centaines de millisecondes ;
• lignes 33-40 : [sendAsync(requête, gestionnaire)] envoie la requête et rend aussitôt un [CompletableFuture<HttpResponse<String>>] ; le gestionnaire de corps [BodyHandlers.ofString()] indique que le corps de la réponse doit être lu comme une chaîne. [thenApply(HttpResponse::body)] n'en garde que le corps. Les six requêtes sont envoyées en même temps et leurs réponses arrivent ensemble au bout d'une seconde (ligne 2 des résultats). Pendant l'attente, aucun thread n'attend la réponse : le client surveille ses connexions avec un thread de sélection (« selector ») ;
• lignes 42-55 : [send] est bloquante ; on l'appelle depuis six threads virtuels. Le résultat est le même : une seconde pour les six requêtes (ligne 3), et aucun thread du système bloqué ;
• lignes 61-66 : la requête GET. [URLEncoder.encode] code les caractères qui ne peuvent pas figurer tels quels dans une adresse (le [É] d'Évry), et le serveur les décode avec [URLDecoder.decode] (ligne 17).

9.9. Les flux de valeurs asynchrones

Une opération asynchrone rend un résultat, au bout d'un certain temps. Un flux ([Stream]) ou un itérateur rend une suite de valeurs, produites à la demande. Il arrive que l'on doive combiner les deux : une suite de valeurs dont chacune demande une attente avant d'être produite. C'est le cas des lignes d'un gros fichier, des lignes du résultat d'une requête SQL, des messages arrivant sur une connexion réseau, des mesures d'un capteur… D'autres langages ont pour cela un type dédié ; Java propose deux modèles :

• le modèle tiré (« pull ») : le consommateur demande la valeur suivante et attend qu'elle soit là. Avec les threads virtuels, c'est simplement un [Iterator] ou un [Stream] dont la production de chaque valeur bloque, lu par un thread virtuel ;
• le modèle poussé (« push ») : le producteur envoie les valeurs à ses abonnés au fur et à mesure qu'elles sont produites, et les abonnés règlent le débit. C'est le modèle des flux réactifs, normalisé par les interfaces [java.util.concurrent.Flow] (Java 9).

9.9.1. Le modèle tiré : un Stream lu par un thread virtuel

Le programme [Async19.java] produit des mesures, une toutes les 200 ms, sous la forme d'un [Stream<Mesure>] :

// une suite de valeurs produites avec des attentes : un Stream « bloquant » lu par un thread virtuel
long debut;

void main() throws InterruptedException {
    // tout le travail est fait par un thread virtuel : ses attentes ne bloquent aucun thread porteur
    Thread.ofVirtual().start(this::consommer).join();
}

void consommer() {
    debut = System.nanoTime();
    // 1 - chaque valeur est traitée dès qu'elle est produite
    try (Stream<Mesure> mesures = lireMesures("capteur-1", 4)) {
        mesures.forEach(mesure -> afficher("1 - reçu " + mesure));
    }

    // 2 - on peut quitter la boucle avant la fin : le producteur ne produit pas les valeurs suivantes
    debut = System.nanoTime();
    try (Stream<Mesure> mesures = lireMesures("capteur-2", 100)) {
        Iterator<Mesure> iterateur = mesures.iterator();
        while (iterateur.hasNext()) {
            Mesure mesure = iterateur.next();
            afficher("2 - reçu " + mesure);
            if (mesure.valeur() >= 22) {
                break;
            }
        }
    }

    // 3 - annulation : un autre thread interrompt le thread qui lit le flux
    debut = System.nanoTime();
    Thread lecteur = Thread.ofVirtual().start(() -> {
        try (Stream<Mesure> mesures = lireMesures("capteur-3", 100)) {
            mesures.forEach(mesure -> afficher("3 - reçu " + mesure));
        } catch (CancellationException e) {
            afficher("3 - lecture annulée");
        }
    });
    attendreFin(lecteur, 700);

    // 4 - toutes les opérations de l'API Stream s'appliquent au flux
    debut = System.nanoTime();
    String fortes;
    try (Stream<Mesure> mesures = lireMesures("capteur-4", 6)) {
        fortes = mesures
                .filter(mesure -> mesure.valeur() > 21)
                .map(mesure -> NumberFormat.getInstance().format(mesure.valeur()))
                .collect(Collectors.joining(" ; ", "[", "]"));
    }
    afficher("4 - mesures supérieures à 21 : " + fortes);
}

// un producteur : une mesure toutes les 200 ms, produite seulement quand le consommateur la demande
Stream<Mesure> lireMesures(String capteur, int nombre) {
    return IntStream.rangeClosed(1, nombre)
            .mapToObj(i -> {
                // l'attente de la mesure suivante
                dormir(200);
                return new Mesure(capteur, i, 19.5 + i * 0.75);
            })
            // exécuté par close() : à la fin normale, après un break ou sur annulation
            .onClose(() -> IO.println("          " + capteur + " : fin de la production"));
}

// attend [delai] ms la fin d'un thread, puis l'interrompt et attend sa fin
void attendreFin(Thread thread, long delai) {
    try {
        thread.join(delai);
        thread.interrupt();
        thread.join();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

// Thread.sleep sans exception contrôlée : une interruption devient une CancellationException
void dormir(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("attente interrompue");
    }
}

// affichage avec le temps écoulé depuis [debut]
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

// une mesure
record Mesure(String capteur, int numero, double valeur) {
    @Override
    public String toString() {
        return capteur + " n° " + numero + " : " + NumberFormat.getInstance().format(valeur) + " °C";
    }
}

Résultats :

> java Async19.java
  249 ms : 1 - reçu capteur-1 n° 1 : 20,25 °C
  455 ms : 1 - reçu capteur-1 n° 2 : 21 °C
  656 ms : 1 - reçu capteur-1 n° 3 : 21,75 °C
  857 ms : 1 - reçu capteur-1 n° 4 : 22,5 °C
          capteur-1 : fin de la production
  200 ms : 2 - reçu capteur-2 n° 1 : 20,25 °C
  401 ms : 2 - reçu capteur-2 n° 2 : 21 °C
  603 ms : 2 - reçu capteur-2 n° 3 : 21,75 °C
  804 ms : 2 - reçu capteur-2 n° 4 : 22,5 °C
          capteur-2 : fin de la production
  201 ms : 3 - reçu capteur-3 n° 1 : 20,25 °C
  402 ms : 3 - reçu capteur-3 n° 2 : 21 °C
  602 ms : 3 - reçu capteur-3 n° 3 : 21,75 °C
          capteur-3 : fin de la production
  702 ms : 3 - lecture annulée
          capteur-4 : fin de la production
 1206 ms : 4 - mesures supérieures à 21 : [21,75 ; 22,5 ; 23,25 ; 24]

Commentaires :

• lignes 52-62 : le producteur. Il rend un [Stream<Mesure>] construit à partir des entiers 1 à [nombre] ; la fonction de [mapToObj] attend 200 ms puis rend une mesure. Un flux étant paresseux, cette fonction n'est appelée que lorsque le consommateur demande la valeur suivante, et chaque mesure est traitée avant que la suivante ne soit produite. [onClose] enregistre une action exécutée à la fermeture du flux, quelle qu'en soit la raison : c'est le [finally] du producteur ;
• ligne 6 : tout le travail est fait par un thread virtuel : les 200 ms d'attente de chaque mesure ne bloquent aucun thread du système ;
• lignes 11-14 : chaque mesure est traitée dès qu'elle est produite : une toutes les 200 ms (lignes 2-5 des résultats). Le flux est déclaré dans un [try] avec ressource : sa fermeture déclenche l'action [onClose] (ligne 6) ;
• lignes 16-27 : pour pouvoir quitter la boucle avant la fin, on parcourt le flux avec son itérateur. Après le [break], le producteur ne produit pas les 96 mesures restantes, et le flux est fermé à la sortie du [try] (ligne 11). Les méthodes [takeWhile] (Java 9) ou [limit] permettent aussi d'arrêter un flux ;
• lignes 29-38 : l'annulation. Le flux est lu par un autre thread virtuel, que la méthode [attendreFin] (lignes 64-73) interrompt au bout de 700 ms. L'attente de la quatrième mesure est interrompue : la méthode [dormir] lance une [CancellationException], qui traverse le flux ; le [try] avec ressource ferme le flux (ligne 15 des résultats), puis le [catch] affiche l'annulation (ligne 16) ;
• lignes 40-49 : tout ce que l'API Stream sait faire s'applique au flux : filtrage, transformation, collecte. Le résultat n'est disponible qu'à la fin du flux, au bout de 6 × 200 ms ;
• lignes 90-96 : la méthode [toString] du record [Mesure] est redéfinie pour un affichage plus compact ; [NumberFormat.getInstance()] formate le réel selon la langue de la machine, sans décimales inutiles ([21], [21,75]).

Le même schéma s'applique à un [BufferedReader] (ses lignes sont lues à la demande, et sa méthode [lines()] rend un [Stream<String>]), à un [ResultSet] JDBC, ou au flux d'entrée d'un socket : sur un thread virtuel, un flux bloquant est un flux asynchrone.

9.9.2. Le modèle poussé : Flow et SubmissionPublisher

L'interface [java.util.concurrent.Flow] (Java 9) regroupe quatre interfaces internes qui normalisent les flux réactifs (« Reactive Streams ») :

• [Flow.Publisher<T>], l'éditeur, produit des valeurs ; sa méthode [subscribe] lui ajoute un abonné ;
• [Flow.Subscriber<T>], l'abonné, reçoit les valeurs par sa méthode [onNext], la fin du flux par [onComplete], une erreur par [onError], et au début l'objet [Flow.Subscription] qui le lie à l'éditeur par [onSubscribe] ;
• [Flow.Subscription], l'abonnement : [request(n)] demande [n] valeurs supplémentaires, [cancel()] met fin à l'abonnement. L'éditeur n'envoie jamais plus de valeurs que l'abonné n'en a demandé : c'est la contre-pression (« backpressure »), qui empêche un éditeur rapide de submerger un abonné lent ;
• [Flow.Processor<T, R>] est à la fois abonné et éditeur (un maillon de transformation).

Java fournit une implémentation de l'éditeur, [SubmissionPublisher<T>], qui livre les valeurs à chaque abonné par l'intermédiaire d'un exécuteur et d'un tampon propre à cet abonné. Les bibliothèques réactives (Project Reactor, RxJava) implémentent ces mêmes interfaces et y ajoutent des dizaines d'opérateurs. Le programme [Async20.java] publie six mesures à deux abonnés, un écran rapide qui se désabonne au-delà de 22 °C et une archive lente :

// le modèle « poussé » (push) : java.util.concurrent.Flow et SubmissionPublisher (Java 9)
final long debut = System.nanoTime();

void main() throws InterruptedException {
    Abonne ecran = new Abonne("écran", 0, 22);
    Abonne archive = new Abonne("archive", 500, Double.MAX_VALUE);
    // l'éditeur confie la livraison des valeurs à un exécuteur et tient un tampon de 2 valeurs par abonné
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        var editeur = new SubmissionPublisher<Mesure>(executeur, 2);
        editeur.subscribe(ecran);
        editeur.subscribe(archive);
        for (int i = 1; i <= 6; i++) {
            dormir(200);
            // submit attend s'il n'y a plus de place dans le tampon d'un abonné ; il rend le retard estimé
            int retard = editeur.submit(new Mesure("capteur-1", i, 19.5 + i * 0.75));
            afficher("éditeur : mesure n° " + i + " publiée, retard estimé : " + retard);
        }
        // close() signale la fin du flux aux abonnés : leur méthode onComplete sera appelée
        afficher("éditeur : fin de la publication");
        editeur.close();
        // on attend que l'archive ait tout traité
        archive.fin.await();
    }
}

// un abonné qui traite chaque mesure en [duree] ms et se désabonne au-delà de [seuil]
class Abonne implements Flow.Subscriber<Mesure> {
    private final String nom;
    private final long duree;
    private final double seuil;
    private Flow.Subscription abonnement;
    final CountDownLatch fin = new CountDownLatch(1);

    Abonne(String nom, long duree, double seuil) {
        this.nom = nom;
        this.duree = duree;
        this.seuil = seuil;
    }

    @Override
    public void onSubscribe(Flow.Subscription abonnement) {
        this.abonnement = abonnement;
        // l'abonné demande les valeurs une par une : c'est lui qui règle le débit
        abonnement.request(1);
    }

    @Override
    public void onNext(Mesure mesure) {
        afficher("  " + nom + " : reçu " + mesure);
        dormir(duree);
        if (mesure.valeur() >= seuil) {
            afficher("  " + nom + " : seuil atteint, désabonnement");
            abonnement.cancel();
        } else {
            abonnement.request(1);
        }
    }

    @Override
    public void onError(Throwable e) {
        afficher("  " + nom + " : erreur " + e);
        fin.countDown();
    }

    @Override
    public void onComplete() {
        afficher("  " + nom + " : fin du flux");
        fin.countDown();
    }
}

// Thread.sleep sans exception contrôlée : une interruption devient une CancellationException
void dormir(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("attente interrompue");
    }
}

// affichage avec le temps écoulé depuis le début
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

// une mesure
record Mesure(String capteur, int numero, double valeur) {
    @Override
    public String toString() {
        return capteur + " n° " + numero + " : " + NumberFormat.getInstance().format(valeur) + " °C";
    }
}

Résultats :

> java Async20.java
  221 ms : éditeur : mesure n° 1 publiée, retard estimé : 1
  269 ms :   archive : reçu capteur-1 n° 1 : 20,25 °C
  274 ms :   écran : reçu capteur-1 n° 1 : 20,25 °C
  466 ms : éditeur : mesure n° 2 publiée, retard estimé : 2
  466 ms :   écran : reçu capteur-1 n° 2 : 21 °C
  667 ms :   écran : reçu capteur-1 n° 3 : 21,75 °C
  667 ms : éditeur : mesure n° 3 publiée, retard estimé : 3
  770 ms :   archive : reçu capteur-1 n° 2 : 21 °C
  867 ms : éditeur : mesure n° 4 publiée, retard estimé : 3
  868 ms :   écran : reçu capteur-1 n° 4 : 22,5 °C
  868 ms :   écran : seuil atteint, désabonnement
 1271 ms :   archive : reçu capteur-1 n° 3 : 21,75 °C
 1271 ms : éditeur : mesure n° 5 publiée, retard estimé : 3
 1771 ms :   archive : reçu capteur-1 n° 4 : 22,5 °C
 1772 ms : éditeur : mesure n° 6 publiée, retard estimé : 3
 1772 ms : éditeur : fin de la publication
 2274 ms :   archive : reçu capteur-1 n° 5 : 23,25 °C
 2775 ms :   archive : reçu capteur-1 n° 6 : 24 °C
 3276 ms :   archive : fin du flux

Commentaires :

• lignes 26-70 : l'abonné, qui traite chaque mesure en [duree] millisecondes et se désabonne lorsque la mesure atteint [seuil] :
  • lignes 40-45 : [onSubscribe] reçoit l'abonnement et demande une première valeur ;
  • lignes 47-57 : [onNext] traite une valeur puis, soit en demande une autre ([request(1)]), soit met fin à l'abonnement ([cancel()]). En demandant les valeurs une par une, l'abonné règle lui-même le débit ;
  • lignes 59-69 : [onError] et [onComplete] signalent la fin du flux, par une erreur ou normalement. Le [CountDownLatch] permet au programme principal d'attendre cette fin ;
• ligne 9 : l'éditeur livre les valeurs à chaque abonné par une tâche de l'exécuteur indiqué, ici des threads virtuels (sans exécuteur, il utiliserait le pool commun, où l'attente de l'archive bloquerait un de ses rares threads), et tient pour chaque abonné un tampon de 2 valeurs au plus ;
• lignes 12-17 : l'éditeur publie une mesure toutes les 200 ms avec [submit], qui rend une estimation du retard : le plus grand nombre de valeurs publiées mais pas encore consommées par un abonné. Lorsque le tampon d'un abonné est plein, [submit] attend qu'une place se libère : la mesure n° 5, prête vers 1 070 ms, n'est publiée qu'à 1 271 ms (ligne 14 des résultats), lorsque l'archive a pris la mesure n° 3. C'est la contre-pression : l'éditeur est ralenti au rythme de l'abonné le plus lent. ([SubmissionPublisher] propose aussi [offer], qui permet d'abandonner une valeur au lieu d'attendre) ;
• les résultats montrent que l'écran, rapide, traite chaque mesure dès sa publication et se désabonne à la mesure n° 4 (22,5 °C, lignes 11-12), tandis que l'archive, qui met 500 ms par mesure, prend du retard et reçoit les six mesures ;
• lignes 18-22 : [close()] signale la fin du flux : chaque abonné encore inscrit recevra [onComplete] après avoir traité les valeurs de son tampon (ligne 20). Le programme attend la fin de l'archive avec le [CountDownLatch].

Le modèle poussé convient aux sources qui produisent à leur rythme (capteurs, événements, messages) et aux chaînes de traitement réactives ; le modèle tiré, plus simple à écrire et à lire, convient à la lecture d'une source à la demande. Avec les threads virtuels, le modèle tiré est le plus souvent suffisant.

9.10. Mettre en cache, revenir sur le bon thread, terminer soi-même

9.10.1. Un cache d'opérations asynchrones

Un [CompletableFuture] est un objet : chaque appel d'une méthode qui en rend un alloue au moins un objet dans le tas, que le ramasse-miettes devra récupérer. C'est négligeable pour une E/S qui dure des millisecondes, et les ramasse-miettes actuels de la JVM traitent très efficacement les objets à durée de vie courte : Java n'a pas de variante « allégée » de [CompletableFuture] (d'autres langages en ont une, pour les méthodes asynchrones appelées des millions de fois qui ont le plus souvent déjà leur résultat).

Pour une méthode asynchrone qui a souvent déjà le résultat, la bonne pratique est de mettre en cache les [CompletableFuture] eux-mêmes : une valeur déjà connue est rendue par un [CompletableFuture] déjà terminé, sans nouvelle allocation, et plusieurs demandes simultanées de la même valeur partagent la même opération en cours. Le programme [Async21.java] l'applique à un service de taux de change :

// un cache de CompletableFuture : une opération asynchrone qui, le plus souvent, a déjà le résultat
void main() {
    var service = new ServiceTaux();
    // 1 - des lectures successives : seule la première lecture d'une devise interroge le serveur
    for (String code : List.of("EUR", "USD", "EUR", "EUR", "USD", "GBP", "EUR")) {
        long debutLecture = System.nanoTime();
        BigDecimal taux = service.lireTaux(code).join();
        IO.println(String.format("%s : %6s lu en %3d ms", code, NumberFormat.getInstance().format(taux),
                (System.nanoTime() - debutLecture) / 1_000_000));
    }
    IO.println("1 - accès au serveur : " + service.nbAccesServeur() + ", lectures : " + service.nbLectures());

    // 2 - trois lectures simultanées de la même devise : elles partagent la même opération
    List<CompletableFuture<BigDecimal>> lectures = List.of(
            service.lireTaux("CHF"), service.lireTaux("CHF"), service.lireTaux("CHF"));
    IO.println("2 - même objet : " + (lectures.get(0) == lectures.get(2))
            + ", taux : " + NumberFormat.getInstance().format(lectures.get(0).join())
            + ", accès au serveur : " + service.nbAccesServeur() + ", lectures : " + service.nbLectures());

    // 3 - une opération en échec n'est pas gardée dans le cache
    try {
        service.lireTaux("XYZ").join();
    } catch (CompletionException e) {
        IO.println("3 - " + e.getCause().getMessage() + ", devises en cache : " + service.devisesEnCache());
    }
}

class ServiceTaux {
    // le cache contient les CompletableFuture eux-mêmes, terminés ou en cours
    private final Map<String, CompletableFuture<BigDecimal>> cache = new ConcurrentHashMap<>();
    private final AtomicInteger nbAccesServeur = new AtomicInteger();
    private final AtomicInteger nbLectures = new AtomicInteger();

    CompletableFuture<BigDecimal> lireTaux(String code) {
        nbLectures.incrementAndGet();
        // si la devise est dans le cache, on rend le CompletableFuture existant : aucun objet n'est créé
        return cache.computeIfAbsent(code, this::lireTauxServeur);
    }

    private CompletableFuture<BigDecimal> lireTauxServeur(String code) {
        nbAccesServeur.incrementAndGet();
        // simule l'interrogation d'un serveur distant (300 ms)
        CompletableFuture<BigDecimal> lecture = new CompletableFuture<Void>()
                .completeOnTimeout(null, 300, TimeUnit.MILLISECONDS)
                .thenApply(rien -> switch (code) {
                    case "EUR" -> BigDecimal.ONE;
                    case "USD" -> new BigDecimal("1.17");
                    case "GBP" -> new BigDecimal("0.87");
                    case "CHF" -> new BigDecimal("0.94");
                    default -> throw new IllegalArgumentException("devise inconnue : " + code);
                });
        // en cas d'échec, on retire l'opération du cache pour qu'une lecture ultérieure soit retentée
        return lecture.whenComplete((taux, e) -> {
            if (e != null) {
                cache.remove(code);
            }
        });
    }

    int nbAccesServeur() {
        return nbAccesServeur.get();
    }

    int nbLectures() {
        return nbLectures.get();
    }

    Set<String> devisesEnCache() {
        return new TreeSet<>(cache.keySet());
    }
}

Résultats :

> java Async21.java
EUR :      1 lu en 363 ms
USD :   1,17 lu en 300 ms
EUR :      1 lu en   0 ms
EUR :      1 lu en   0 ms
USD :   1,17 lu en   0 ms
GBP :   0,87 lu en 300 ms
EUR :      1 lu en   0 ms
1 - accès au serveur : 3, lectures : 7
2 - même objet : true, taux : 0,94, accès au serveur : 4, lectures : 10
3 - devise inconnue : XYZ, devises en cache : [CHF, EUR, GBP, USD]

Commentaires :

• ligne 30 : le cache associe à chaque code de devise le [CompletableFuture] de son taux, terminé ou encore en cours. C'est une [ConcurrentHashMap] (chapitre 8), car il peut être utilisé par plusieurs threads à la fois (le thread principal, le thread de la minuterie qui termine les lectures) ;
• lignes 34-38 : [computeIfAbsent(clé, fonction)] rend la valeur associée à la clé si elle existe ; sinon, elle appelle la fonction, range son résultat dans le dictionnaire et le rend. Pour une [ConcurrentHashMap], cette opération est atomique : deux threads qui demandent en même temps une devise absente ne déclenchent qu'un seul accès au serveur ;
• lignes 40-58 : l'interrogation simulée du serveur (300 ms). La continuation [thenApply] contient une expression [switch] sur une chaîne, dont la branche [default] lance une exception pour une devise inconnue ;
• lignes 52-57 : en cas d'échec, on retire l'opération du cache, pour qu'une demande ultérieure de la même devise interroge à nouveau le serveur au lieu de rendre indéfiniment le même échec. C'est le [CompletableFuture] rendu par [whenComplete] qui est rangé dans le cache : il ne se termine qu'après l'exécution de l'action, ce qui garantit à l'appelant que le cache est à jour quand il reçoit le résultat ;
• lignes 5-11 : pour l'appelant, rien ne change : il attend le résultat avec [join()]. Les résultats montrent les 3 accès au serveur (300 ms chacun ; un peu plus pour le premier, pendant lequel la JVM charge les classes nécessaires) et les 4 réponses immédiates du cache ;
• lignes 13-18 : trois demandes simultanées du taux du franc suisse obtiennent le même objet et ne provoquent qu'un accès au serveur (ligne 10 des résultats) ;
• lignes 20-25 : la devise inconnue a provoqué un échec et n'est plus dans le cache (ligne 11).

9.10.2. Revenir sur le thread de l'interface graphique

Le programme [Async06.java] a montré qu'une continuation est exécutée, selon le cas, par le thread qui termine l'opération ou par un thread de l'exécuteur choisi. Dans une application graphique (Swing, JavaFX), cela compte : un composant graphique ne peut être manipulé que par le thread de l'interface. Java n'a pas de mécanisme qui ramène automatiquement la suite du code sur ce thread : c'est au programme de le demander, en passant le bon exécuteur à la continuation, ou en utilisant les outils de la bibliothèque graphique :

• avec Swing, [SwingUtilities.invokeLater(action)] fait exécuter l'action par le thread de l'interface ; la référence de méthode [SwingUtilities::invokeLater] peut servir d'exécuteur à une continuation : [telechargement.thenAcceptAsync(page -> zoneTexte.setText(page), SwingUtilities::invokeLater)]. La classe [SwingWorker] propose aussi une méthode [doInBackground], exécutée par un thread d'arrière-plan, et des méthodes [process] et [done], exécutées par le thread de l'interface ;
• avec JavaFX, [Platform.runLater(action)] joue le même rôle, et les classes [Task] et [Service] de [javafx.concurrent] encapsulent un traitement d'arrière-plan avec sa progression.

Dans tous les cas, la règle est la même : le thread de l'interface ne doit jamais attendre ([join()], [get()], E/S bloquante), sans quoi la fenêtre se fige. Le traitement long est exécuté par un autre thread (un thread virtuel, par exemple), et seule la mise à jour finale des composants revient au thread de l'interface.

9.10.3. Terminer soi-même un CompletableFuture

Il arrive que l'on doive utiliser une API qui signale ses événements par des écouteurs (« listeners », chapitre 7), alors que le reste du code est écrit avec des [CompletableFuture]. On a vu avec [Async17.java] la technique : on crée un [CompletableFuture] « en cours » avec son constructeur, on le rend aussitôt, et on le termine soi-même, au moment voulu :

• [complete(valeur)] le termine avec succès ;
• [completeExceptionally(exception)] le termine en échec ;
• [cancel(false)] l'annule.

Ces trois méthodes sont sans effet si le [CompletableFuture] est déjà terminé ([complete] et [completeExceptionally] rendent alors [false]) : le premier qui le termine l'emporte, ce qui règle naturellement les situations de concurrence (un résultat et un délai qui arrivent au même moment, par exemple). Le programme [Async22.java] attend qu'un thermomètre, qui notifie ses écouteurs à chaque mesure, atteigne une température :

// un CompletableFuture terminé « à la main » : transformer une API à base d'écouteurs en opération asynchrone
final long debut = System.nanoTime();

void main() {
    var thermometre = new Thermometre();
    thermometre.demarrer();

    // 1 - on attend la première mesure supérieure ou égale à 21 °C
    double mesure = attendreTemperature(thermometre, 21).join();
    afficher("1 - température atteinte : " + format(mesure) + " °C");

    // 2 - la même attente, limitée à une seconde
    try {
        mesure = attendreTemperature(thermometre, 30).orTimeout(1, TimeUnit.SECONDS).join();
        afficher("2 - température atteinte : " + format(mesure) + " °C");
    } catch (CompletionException e) {
        afficher("2 - 30 °C non atteints en une seconde : " + e.getCause().getClass().getSimpleName());
    }
    thermometre.arreter();
}

// transforme les notifications du thermomètre en un CompletableFuture
CompletableFuture<Double> attendreTemperature(Thermometre thermometre, double seuil) {
    // un CompletableFuture que l'on terminera soi-même
    CompletableFuture<Double> resultat = new CompletableFuture<>();
    // l'écouteur termine le CompletableFuture lorsque le seuil est atteint
    DoubleConsumer ecouteur = temperature -> {
        if (temperature >= seuil) {
            resultat.complete(temperature);
        }
    };
    thermometre.ajouterEcouteur(ecouteur);
    // quelle que soit la façon dont il se termine (seuil, délai, annulation), on retire l'écouteur
    resultat.whenComplete((temperature, e) -> thermometre.retirerEcouteur(ecouteur));
    // le CompletableFuture est rendu tout de suite ; il sera terminé plus tard
    return resultat;
}

// un thermomètre qui notifie ses écouteurs toutes les 200 ms (une API « à l'ancienne »)
class Thermometre {
    private final List<DoubleConsumer> ecouteurs = new CopyOnWriteArrayList<>();
    private final ScheduledExecutorService minuterie = Executors.newSingleThreadScheduledExecutor();
    private double temperature = 19;

    void ajouterEcouteur(DoubleConsumer ecouteur) {
        ecouteurs.add(ecouteur);
    }

    void retirerEcouteur(DoubleConsumer ecouteur) {
        ecouteurs.remove(ecouteur);
    }

    void demarrer() {
        minuterie.scheduleAtFixedRate(() -> {
            temperature += 0.5;
            IO.println("          mesure : " + format(temperature) + " °C");
            ecouteurs.forEach(ecouteur -> ecouteur.accept(temperature));
        }, 200, 200, TimeUnit.MILLISECONDS);
    }

    void arreter() {
        minuterie.shutdown();
    }
}

// un réel au format français, sans décimales inutiles
String format(double valeur) {
    return NumberFormat.getInstance().format(valeur);
}

// affichage avec le temps écoulé depuis le début
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async22.java
          mesure : 19,5 °C
          mesure : 20 °C
          mesure : 20,5 °C
          mesure : 21 °C
  807 ms : 1 - température atteinte : 21 °C
          mesure : 21,5 °C
          mesure : 22 °C
          mesure : 22,5 °C
          mesure : 23 °C
          mesure : 23,5 °C
 1809 ms : 2 - 30 °C non atteints en une seconde : TimeoutException

Commentaires :

• lignes 39-64 : le thermomètre, une classe « à l'ancienne » : un [ScheduledExecutorService] augmente la température de 0,5 °C toutes les 200 ms et la transmet à ses écouteurs, de type [DoubleConsumer] (un [Consumer] spécialisé pour le type [double]). La liste des écouteurs est une [CopyOnWriteArrayList], une liste qui peut être modifiée par un thread (ajout ou retrait d'un écouteur) pendant qu'un autre la parcourt ;
• lignes 22-37 : la méthode qui transforme les notifications en [CompletableFuture]. Elle crée un [CompletableFuture<Double>] en cours, inscrit un écouteur auprès du thermomètre et rend aussitôt le [CompletableFuture], qui n'est pas terminé ;
• lignes 27-31 : à chaque mesure, l'écouteur teste le seuil. Lorsqu'il est atteint, il termine le [CompletableFuture] avec la température comme résultat. Les mesures suivantes appellent encore [complete], sans effet, jusqu'au retrait de l'écouteur ;
• ligne 34 : quelle que soit la façon dont le [CompletableFuture] se termine (seuil atteint, délai dépassé, annulation par l'appelant), [whenComplete] retire l'écouteur. Sans cela, un écouteur dont le seuil n'est jamais atteint resterait inscrit pour toujours ;
• lignes 8-18 : pour l'appelant, l'attente d'une température est une opération asynchrone comme les autres : il l'attend avec [join()], il la limite dans le temps avec [orTimeout]. Comme [orTimeout] termine le [CompletableFuture] lui-même (et non une copie), l'action [whenComplete] de la ligne 34 est exécutée et l'écouteur est retiré.

Une précaution : les continuations attachées au [CompletableFuture] sont exécutées par le thread qui appelle [complete], ici le thread du thermomètre, au milieu de la notification de ses écouteurs. Si ces continuations sont longues, on les écrit avec les variantes [xxxAsync] et un exécuteur, ou l'on termine le [CompletableFuture] par [completeAsync(() -> valeur, executeur)] (Java 9), qui fait appeler [complete] par l'exécuteur.

9.11. Le parallélisme limité

9.11.1. Threads virtuels et Semaphore

[allOf] et un thread virtuel par opération lancent toutes les opérations en même temps. Ce n'est pas toujours souhaitable : pour télécharger 10 000 pages web, on ne veut pas ouvrir 10 000 connexions simultanées vers le même serveur. Avec des threads de plate-forme, on limitait la concurrence par la taille du pool de threads. Avec les threads virtuels, on ne met pas les threads en pool (ils sont si peu coûteux qu'on en crée un par tâche) : on limite l'accès à la ressource rare avec un [Semaphore] (chapitre 8), dont les autorisations représentent le nombre d'opérations simultanées permises. Le programme [Async23.java] télécharge neuf pages, trois à la fois au plus :

// traiter une collection avec un nombre limité d'opérations simultanées : threads virtuels et Semaphore
final AtomicInteger enCours = new AtomicInteger();

long debut;

void main() {
    debut = System.nanoTime();
    List<String> pages = IntStream.rangeClosed(1, 9).mapToObj(i -> "page" + i + ".html").toList();
    // au plus 3 téléchargements en même temps : un sémaphore à 3 autorisations
    var autorisations = new Semaphore(3);
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        // un thread virtuel par page : ils sont bon marché, on peut en créer autant qu'il y a de pages
        for (String page : pages) {
            executeur.submit(() -> {
                // le thread attend une autorisation (sans bloquer de thread porteur)
                autorisations.acquire();
                try {
                    telecharger(page);
                } finally {
                    autorisations.release();
                }
                return null;
            });
        }
    } // close() attend la fin de tous les téléchargements
    afficher("les " + pages.size() + " pages sont téléchargées");
}

void telecharger(String page) throws InterruptedException {
    // compteur partagé par les opérations simultanées : on le modifie de façon atomique
    int n = enCours.incrementAndGet();
    afficher("début de " + page + " (" + n + " en cours)");
    // le téléchargement simulé d'une page
    Thread.sleep(1000);
    enCours.decrementAndGet();
    afficher("fin de " + page);
}

// affichage avec le temps écoulé depuis le début et le thread qui exécute l'affichage
void afficher(String message) {
    Thread thread = Thread.currentThread();
    String nom = thread.isVirtual() ? "virtuel #" + thread.threadId() : thread.getName();
    IO.println(String.format("%5d ms [%-11s] : %s", (System.nanoTime() - debut) / 1_000_000, nom, message));
}

Résultats :

> java Async23.java
   13 ms [virtuel #25] : début de page2.html (2 en cours)
   13 ms [virtuel #27] : début de page3.html (3 en cours)
   13 ms [virtuel #23] : début de page1.html (1 en cours)
 1026 ms [virtuel #25] : fin de page2.html
 1026 ms [virtuel #27] : fin de page3.html
 1026 ms [virtuel #23] : fin de page1.html
 1026 ms [virtuel #28] : début de page4.html (1 en cours)
 1027 ms [virtuel #29] : début de page5.html (2 en cours)
 1027 ms [virtuel #30] : début de page6.html (3 en cours)
 2027 ms [virtuel #29] : fin de page5.html
 2027 ms [virtuel #30] : fin de page6.html
 2028 ms [virtuel #31] : début de page7.html (1 en cours)
 2028 ms [virtuel #32] : début de page8.html (2 en cours)
 2027 ms [virtuel #28] : fin de page4.html
 2028 ms [virtuel #33] : début de page9.html (3 en cours)
 3028 ms [virtuel #31] : fin de page7.html
 3028 ms [virtuel #32] : fin de page8.html
 3029 ms [virtuel #33] : fin de page9.html
 3030 ms [main       ] : les 9 pages sont téléchargées

Commentaires :

• ligne 10 : un sémaphore à 3 autorisations ;
• lignes 13-24 : un thread virtuel par page. La tâche est une lambda qui se termine par [return null;] : c'est donc un [Callable], dont la méthode peut lancer des exceptions contrôlées, ce qui évite de traiter ici l'[InterruptedException] de [acquire] et de [Thread.sleep] ;
• lignes 16-21 : chaque thread attend une autorisation ([acquire]), télécharge sa page puis rend l'autorisation ([release]) dans un [finally], pour qu'elle soit rendue même si le téléchargement échoue. Un thread virtuel qui attend une autorisation est décroché de son thread porteur, comme pendant une E/S ;
• lignes 29-37 : le compteur [enCours], partagé par les threads, est un [AtomicInteger] modifié de façon atomique ;
• les résultats montrent les trois vagues de trois pages : dès qu'une page est terminée, une autre commence, et le nombre de téléchargements en cours ne dépasse jamais 3. L'ensemble dure 3 secondes au lieu de 9. Lorsque deux threads écrivent presque au même instant, l'ordre des lignes peut varier d'une exécution à l'autre, et un instant affiché peut même être légèrement inférieur à celui de la ligne précédente ;
• chaque page a son thread virtuel (n° 23 à 33 ; les numéros sont communs à tous les threads de la JVM, threads porteurs compris, d'où les trous), mais ces neuf threads virtuels ont été exécutés par les deux seuls threads porteurs de la machine.

9.11.2. Producteur / consommateurs : BlockingQueue

Le chapitre précédent a présenté le schéma producteur / consommateur : des producteurs déposent des éléments dans une file, des consommateurs les retirent. L'interface [BlockingQueue<E>] en est l'outil : [put] dépose un élément en attendant, si la file est bornée et pleine, qu'une place se libère ; [take] retire un élément en attendant, si la file est vide, qu'il en arrive un ; [offer] et [poll] tentent la même chose sans attendre (ou avec un délai). Ces attentes sont bloquantes ; exécutées par des threads virtuels, elles n'immobilisent aucun thread du système, et la file bloquante devient l'outil de communication entre opérations asynchrones.

Le programme [Async24.java] fait traiter des commandes, produites toutes les 50 ms, par deux consommateurs qui mettent 400 ms à traiter chacune d'elles :

// producteur / consommateurs avec une file bloquante, exécutés par des threads virtuels
// la valeur qui signale aux consommateurs qu'il n'y aura plus de commandes
static final int FIN = -1;
// une file bornée : elle contient au plus 2 commandes en attente de traitement
final BlockingQueue<Integer> file = new ArrayBlockingQueue<>(2);
long debut;

void main() {
    debut = System.nanoTime();
    // le producteur et deux consommateurs travaillent en même temps
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        executeur.submit(() -> produire(8));
        executeur.submit(() -> consommer("C1"));
        executeur.submit(() -> consommer("C2"));
    }
    afficher("toutes les commandes ont été traitées");
}

// le producteur : une commande toutes les 50 ms
Void produire(int nbCommandes) throws InterruptedException {
    for (int i = 1; i <= nbCommandes; i++) {
        Thread.sleep(50);
        // offer dépose la commande tout de suite s'il y a de la place dans la file
        if (file.offer(i)) {
            afficher("P  : commande " + i + " déposée");
        } else {
            // file pleine : put attend qu'une place se libère
            afficher("P  : file pleine, la commande " + i + " attend");
            file.put(i);
            afficher("P  : commande " + i + " déposée");
        }
    }
    // plus rien ne sera produit : on dépose la marque de fin
    file.put(FIN);
    afficher("P  : fin de la production");
    return null;
}

// un consommateur : traiter une commande prend 400 ms
Void consommer(String nom) throws InterruptedException {
    while (true) {
        // take attend qu'il y ait une commande dans la file
        int commande = file.take();
        if (commande == FIN) {
            // on remet la marque de fin dans la file pour l'autre consommateur
            file.put(FIN);
            break;
        }
        afficher(nom + " : traitement de la commande " + commande);
        Thread.sleep(400);
    }
    afficher(nom + " : plus de commandes");
    return null;
}

// affichage avec le temps écoulé depuis le début
void afficher(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - debut) / 1_000_000, message));
}

Résultats :

> java Async24.java
   58 ms : P  : commande 1 déposée
   62 ms : C1 : traitement de la commande 1
  118 ms : C2 : traitement de la commande 2
  119 ms : P  : commande 2 déposée
  169 ms : P  : commande 3 déposée
  220 ms : P  : commande 4 déposée
  271 ms : P  : file pleine, la commande 5 attend
  469 ms : P  : commande 5 déposée
  469 ms : C1 : traitement de la commande 3
  519 ms : C2 : traitement de la commande 4
  519 ms : P  : commande 6 déposée
  570 ms : P  : file pleine, la commande 7 attend
  870 ms : C1 : traitement de la commande 5
  870 ms : P  : commande 7 déposée
  920 ms : C2 : traitement de la commande 6
  920 ms : P  : commande 8 déposée
 1270 ms : C1 : traitement de la commande 7
 1271 ms : P  : fin de la production
 1320 ms : C2 : traitement de la commande 8
 1672 ms : C1 : plus de commandes
 1721 ms : C2 : plus de commandes
 1721 ms : toutes les commandes ont été traitées

Commentaires :

• ligne 5 : une [ArrayBlockingQueue] de capacité 2 : elle contient au plus 2 commandes en attente ;
• lignes 11-15 : le producteur et les deux consommateurs sont exécutés par trois threads virtuels ; le programme attend leur fin à tous les trois par la méthode [close()] de l'exécuteur. Les méthodes [produire] et [consommer] rendent [Void] et se terminent par [return null;] : les lambdas des lignes 12-14 sont ainsi des [Callable], et les méthodes peuvent déclarer [InterruptedException] ;
• lignes 19-37 : le producteur tente de déposer chaque commande avec [offer]. Si la file est pleine, il l'indique et attend une place avec [put] : le producteur est ralenti au rythme des consommateurs (lignes 8-9 des résultats). C'est l'intérêt d'une file bornée : un producteur trop rapide ne peut pas remplir la mémoire ;
• ligne 34 : la production est terminée ; le producteur dépose une valeur particulière, la marque de fin [FIN] (ligne 3), que les consommateurs reconnaîtront. Une file bloquante n'a pas, en effet, de méthode pour signaler qu'elle est « fermée » ;
• lignes 39-54 : chaque consommateur retire les commandes avec [take]. Une commande n'est livrée qu'à un seul consommateur : les commandes se répartissent entre C1 et C2 selon leur disponibilité. Le consommateur qui reçoit la marque de fin la remet dans la file pour l'autre consommateur, puis s'arrête ;
• comme dans l'exemple précédent, deux lignes affichées presque au même instant peuvent apparaître dans le désordre.

9.12. Le temps : ScheduledExecutorService et Clock

Deux outils complètent ceux qui sont liés au temps :

• un [ScheduledExecutorService] (Java 5) est un exécuteur qui sait exécuter une tâche après un délai ([schedule]) ou périodiquement ([scheduleAtFixedRate], à intervalles fixes mesurés d'un début d'exécution au suivant ; [scheduleWithFixedDelay], avec un délai fixe entre la fin d'une exécution et le début de la suivante). Une tâche périodique n'est jamais exécutée deux fois en même temps : si une exécution dure plus longtemps que la période, la suivante est retardée. On obtient un exécuteur à un thread avec [Executors.newSingleThreadScheduledExecutor()] ;
• la classe abstraite [java.time.Clock] (Java 8) représente l'horloge : elle donne l'instant présent et le fuseau horaire. [Clock.systemDefaultZone()] est l'horloge du système dans le fuseau de la machine. Toutes les méthodes [now()] de l'API [java.time] ([LocalDateTime.now(horloge)], [Instant.now(horloge)]…) acceptent une horloge. Une classe qui dépend de l'heure reçoit une [Clock] au lieu d'appeler [LocalDateTime.now()] : on peut alors la tester avec une horloge contrôlée.

Le programme [Async25.java] utilise les deux :

// le temps : ScheduledExecutorService et Clock
void main() throws InterruptedException {
    // 1 - une action toutes les 500 ms, arrêtée au bout de 2,2 secondes
    long debut = System.nanoTime();
    AtomicInteger nbTops = new AtomicInteger();
    try (ScheduledExecutorService planificateur = Executors.newSingleThreadScheduledExecutor()) {
        // scheduleAtFixedRate : premier top dans 500 ms, puis un top toutes les 500 ms
        ScheduledFuture<?> tops = planificateur.scheduleAtFixedRate(
                () -> IO.println(String.format("%6d ms : top n° %d",
                        (System.nanoTime() - debut) / 1_000_000, nbTops.incrementAndGet())),
                500, 500, TimeUnit.MILLISECONDS);
        // schedule : une action unique, dans 2,2 secondes, qui arrête les tops
        ScheduledFuture<?> arret = planificateur.schedule(
                () -> tops.cancel(false), 2200, TimeUnit.MILLISECONDS);
        arret.get();
        IO.println(String.format("%6d ms : minuterie arrêtée après %d tops",
                (System.nanoTime() - debut) / 1_000_000, nbTops.get()));
    } catch (ExecutionException e) {
        IO.println("erreur : " + e.getCause());
    }

    // 2 - Clock : une classe qui dépend de l'heure reçoit son horloge au lieu d'appeler LocalDateTime.now()
    Clock matin = Clock.fixed(Instant.parse("2026-10-03T09:30:00Z"), ZoneOffset.UTC);
    Clock soir = Clock.fixed(Instant.parse("2026-10-03T19:45:00Z"), ZoneOffset.UTC);
    IO.println(new Accueil(matin).saluer("Pierre"));
    IO.println(new Accueil(soir).saluer("Pierre"));
    IO.println(new Accueil(Clock.systemDefaultZone()).saluer("Pierre") + " (horloge du système)");
}

// une classe qui dépend de l'heure courante
record Accueil(Clock horloge) {
    String saluer(String nom) {
        LocalDateTime maintenant = LocalDateTime.now(horloge);
        String formule = maintenant.getHour() < 18 ? "Bonjour" : "Bonsoir";
        String date = maintenant.format(DateTimeFormatter.ofPattern("dd/MM/yyyy HH:mm"));
        return "[" + date + "] " + formule + " " + nom;
    }
}

Résultats :

> java Async25.java
   504 ms : top n° 1
  1003 ms : top n° 2
  1503 ms : top n° 3
  2003 ms : top n° 4
  2204 ms : minuterie arrêtée après 4 tops
[03/10/2026 09:30] Bonjour Pierre
[03/10/2026 19:45] Bonsoir Pierre
[03/10/2026 19:18] Bonsoir Pierre (horloge du système)

Commentaires :

• lignes 8-11 : [scheduleAtFixedRate(tâche, délai initial, période, unité)] exécute la tâche une première fois au bout de 500 ms, puis toutes les 500 ms. Elle rend un [ScheduledFuture], un [Future] qui permet notamment d'annuler les exécutions à venir. Le compteur de tops est un [AtomicInteger] : une lambda ne peut pas modifier une variable locale ;
• lignes 13-15 : [schedule(tâche, délai, unité)] exécute une tâche unique au bout de 2,2 secondes : elle annule les tops. [arret.get()] attend la fin de cette tâche. Les tops ont eu lieu à 0,5, 1, 1,5 et 2 secondes, et la minuterie est arrêtée à 2,2 secondes, après 4 tops ;
• lignes 30-38 : le record [Accueil] dit « Bonjour » ou « Bonsoir » selon l'heure. Il reçoit son horloge dans son constructeur au lieu d'appeler [LocalDateTime.now()] : c'est une dépendance injectée, comme la couche [dao] est injectée dans la couche [metier] au chapitre 6 ;
• lignes 23-27 : [Clock.fixed(instant, fuseau)] rend une horloge arrêtée, qui donne toujours le même instant : elle permet de vérifier le comportement d'[Accueil] le matin et le soir sans attendre le soir. La dernière ligne utilise l'horloge du système (le résultat dépend donc de l'heure d'exécution). Pour des tests plus élaborés, on peut écrire sa propre classe dérivée de [Clock], que l'on fait avancer à volonté. L'interface [InstantSource] (Java 17), qui ne donne que l'instant présent, est une dépendance plus légère encore lorsque le fuseau horaire est inutile.

Pour mesurer une durée, on n'utilise pas l'horloge (qui peut être ajustée pendant la mesure, par une synchronisation de l'heure du système par exemple) mais [System.nanoTime()], comme dans les exemples de ce chapitre.

9.13. La concurrence structurée (aperçu)

Dans les exemples précédents, les opérations lancées par un programme vivent indépendamment de lui : si l'une échoue, les autres continuent ; si le code qui les a lancées est interrompu, elles continuent aussi ; et rien n'oblige ce code à attendre leur fin. La concurrence structurée (« structured concurrency ») propose de traiter un groupe de sous-tâches comme un bloc : elles sont lancées dans une portée, la portée attend leur fin, et si l'une échoue, les autres sont annulées. La classe [StructuredTaskScope] met cette idée en œuvre avec des threads virtuels. Elle est disponible depuis Java 21, mais toujours en aperçu (« preview ») dans Java 25 : son API peut encore changer, et un programme qui l'utilise doit être compilé et exécuté avec l'option [--enable-preview]. Elle ne fait donc pas partie des programmes du cours ; le code suivant (fichier [apercu/ConcurrenceStructuree.txt], à copier dans un fichier [ConcurrenceStructuree.java] pour l'essayer) en donne une idée :

// concurrence structurée : API en aperçu (preview) dans Java 25
// exécution : java --enable-preview ConcurrenceStructuree.java
import java.util.concurrent.StructuredTaskScope.Joiner;
import java.util.concurrent.StructuredTaskScope.Subtask;

void main() throws InterruptedException {
    // 1 - deux sous-tâches qui doivent réussir toutes les deux
    try (var portee = StructuredTaskScope.open()) {
        Subtask<Integer> angers = portee.fork(() -> lireTemperature("Angers", 1000));
        Subtask<Integer> brest = portee.fork(() -> lireTemperature("Brest", 500));
        // join attend la fin des sous-tâches ; si l'une échoue, l'autre est annulée
        // et join lance une exception
        portee.join();
        IO.println("Angers : " + angers.get() + ", Brest : " + brest.get());
    }
    // 2 - la première réponse suffit : les autres sous-tâches sont annulées
    try (var portee = StructuredTaskScope.open(Joiner.<Integer>anySuccessfulResultOrThrow())) {
        portee.fork(() -> lireTemperature("miroir-paris", 800));
        portee.fork(() -> lireTemperature("miroir-lyon", 300));
        IO.println("première réponse : " + portee.join());
    }
}

int lireTemperature(String ville, long duree) throws InterruptedException {
    Thread.sleep(duree);
    return 10 + ville.length();
}

Résultats :

> java --enable-preview ConcurrenceStructuree.java
Angers : 16, Brest : 15
première réponse : 21

Commentaires :

• ligne 8 : [StructuredTaskScope.open()] ouvre une portée, déclarée dans un [try] avec ressource : sa fermeture garantit qu'aucune sous-tâche ne lui survit ;
• lignes 9-10 : [fork] lance une sous-tâche dans un nouveau thread virtuel et rend un [Subtask] ;
• ligne 13 : [join()] attend la fin de toutes les sous-tâches. Avec la politique par défaut, si une sous-tâche échoue, les autres sont annulées (interrompues) et [join()] lance une exception ;
• ligne 14 : les résultats des sous-tâches, toutes terminées ;
• lignes 17-21 : une autre politique, fournie par un [Joiner] : la première sous-tâche qui réussit donne le résultat de [join()], et les autres sont annulées. C'est l'équivalent structuré de [anyOf], mais les requêtes perdantes sont réellement arrêtées.

9.14. Application : le calcul de l'impôt avec des threads virtuels

Le programme [Async26.java] rassemble plusieurs outils du chapitre autour du calcul de l'impôt du fil rouge. Il lit les tranches d'impôt dans le fichier [DataImpot.txt] déjà utilisé, lit une liste de contribuables dans le fichier [contribuables.txt] :

o 2 60000
o 3 60000
n 0 60000
n 2 25000
o 5 120000
x 1 20000
o 1 -5

(chaque ligne contient : marié [o/n], nombre d'enfants, salaire annuel), calcule l'impôt de chacun et écrit les résultats dans le fichier [resultats.txt]. Les deux lectures sont menées de front par des threads virtuels et l'ensemble du traitement est limité à 10 secondes.

// application : le calcul de l'impôt, avec des entrées / sorties exécutées par des threads virtuels
// - les tranches sont lues dans [DataImpot.txt], les contribuables dans [contribuables.txt]
// - les résultats sont écrits dans [resultats.txt]
// une ligne de contribuable correcte : marié (o/n), nombre d'enfants, salaire annuel
static final Pattern CONTRIBUABLE = Pattern.compile("([on])\\s+(\\d+)\\s+(\\d+)");

void main() {
    Path resultats = Path.of("resultats.txt");
    try (ExecutorService executeur = Executors.newVirtualThreadPerTaskExecutor()) {
        // les deux fichiers d'entrée sont lus en même temps, par deux threads virtuels
        Future<List<TrancheImpot>> lectureTranches = executeur.submit(
                () -> chargerTranches(Path.of("DataImpot.txt")));
        Future<List<String>> lectureContribuables = executeur.submit(
                () -> Files.readAllLines(Path.of("contribuables.txt")));
        // le calcul attend les deux lectures, puis écrit les résultats
        Future<Integer> traitement = executeur.submit(
                () -> calculerImpots(lectureTranches.get(), lectureContribuables.get(), resultats));
        try {
            // l'ensemble du traitement est limité à 10 secondes
            int nbResultats = traitement.get(10, TimeUnit.SECONDS);
            IO.println(nbResultats + " lignes écrites dans [" + resultats + "] :");
            IO.print(Files.readString(resultats));
        } catch (TimeoutException e) {
            traitement.cancel(true);
            IO.println("traitement abandonné : délai dépassé");
        }
    } catch (ExecutionException e) {
        // l'erreur d'une lecture est enveloppée deux fois : par la lecture et par le calcul qui l'attendait
        Throwable cause = e.getCause();
        while (cause instanceof ExecutionException) {
            cause = cause.getCause();
        }
        IO.println("erreur : " + cause.getClass().getSimpleName() + " : " + cause.getMessage());
    } catch (IOException | InterruptedException e) {
        IO.println("erreur : " + e.getClass().getSimpleName() + " : " + e.getMessage());
    }
}

// lit les tranches d'impôt (limite:coeffR:coeffN, point décimal)
List<TrancheImpot> chargerTranches(Path fichier) throws IOException {
    try (Stream<String> lignes = Files.lines(fichier)) {
        return lignes.filter(ligne -> !ligne.isBlank()).map(TrancheImpot::parse).toList();
    }
}

// calcule l'impôt de chaque contribuable et écrit les résultats
int calculerImpots(List<TrancheImpot> tranches, List<String> contribuables, Path sortie) throws IOException {
    try (BufferedWriter ecrivain = Files.newBufferedWriter(sortie)) {
        for (String ligne : contribuables) {
            Matcher champs = CONTRIBUABLE.matcher(ligne.trim());
            String resultat = champs.matches()
                    ? "impôt = " + calculerImpot(tranches, champs.group(1).equals("o"),
                            Integer.parseInt(champs.group(2)), Integer.parseInt(champs.group(3))) + " €"
                    : "ligne incorrecte";
            ecrivain.write(String.format("%-12s : %s", ligne, resultat));
            ecrivain.newLine();
        }
    }
    return contribuables.size();
}

// le calcul de l'impôt (barème 2004 sur les revenus 2003, conservé pour l'exercice)
int calculerImpot(List<TrancheImpot> tranches, boolean marie, int nbEnfants, int salaire) {
    BigDecimal nbParts = BigDecimal.valueOf(nbEnfants).divide(BigDecimal.TWO)
            .add(BigDecimal.valueOf(marie ? 2 : 1));
    if (nbEnfants >= 3) {
        nbParts = nbParts.add(new BigDecimal("0.5"));
    }
    BigDecimal revenu = new BigDecimal("0.72").multiply(BigDecimal.valueOf(salaire));
    BigDecimal quotient = revenu.divide(nbParts, MathContext.DECIMAL64);
    // la première tranche dont la limite dépasse le quotient familial, sinon la dernière
    TrancheImpot tranche = tranches.stream()
            .filter(t -> quotient.compareTo(t.limite()) <= 0)
            .findFirst()
            .orElse(tranches.getLast());
    return tranche.coeffR().multiply(revenu).subtract(tranche.coeffN().multiply(nbParts)).intValue();
}

// une tranche d'impôt
record TrancheImpot(BigDecimal limite, BigDecimal coeffR, BigDecimal coeffN) {
    static TrancheImpot parse(String ligne) {
        String[] champs = ligne.split(":");
        return new TrancheImpot(new BigDecimal(champs[0]), new BigDecimal(champs[1]),
                new BigDecimal(champs[2]));
    }
}

Résultats :

> java Async26.java
7 lignes écrites dans [resultats.txt] :
o 2 60000    : impôt = 4282 €
o 3 60000    : impôt = 2959 €
n 0 60000    : impôt = 11519 €
n 2 25000    : impôt = 792 €
o 5 120000   : impôt = 11109 €
x 1 20000    : ligne incorrecte
o 1 -5       : ligne incorrecte

Le même programme, exécuté dans un dossier où le fichier [DataImpot.txt] est absent :

> java Async26.java
erreur : NoSuchFileException : DataImpot.txt

Commentaires :

• lignes 11-14 : les deux fichiers d'entrée sont lus en même temps, par deux threads virtuels, avec des méthodes de lecture bloquantes ordinaires ([Files.lines], [Files.readAllLines]) ;
• lignes 16-17 : le calcul est une troisième tâche, qui commence par attendre les deux lectures avec [get()]. Ces attentes bloquent un thread virtuel, ce qui ne coûte rien ;
• lignes 18-26 : le programme principal attend le résultat au plus 10 secondes. Au-delà, il annule le traitement (ce qui interrompt son thread virtuel) et le signale ;
• lignes 27-36 : les erreurs sont traitées à un seul endroit, dans le programme principal. Une erreur de lecture (fichier absent, nombre mal écrit) est rangée dans le [Future] de la lecture ; le [get()] du calcul la relance dans une [ExecutionException], qui devient à son tour l'exception du calcul, relancée par le [get] du programme principal dans une seconde [ExecutionException]. La boucle des lignes 30-32 remonte à la cause d'origine (second résultat). C'est l'inconvénient des tâches dépendantes ; la concurrence structurée, présentée au paragraphe précédent, le fera disparaître ;
• lignes 39-44 : la lecture des tranches, avec un [Stream] de lignes qui ignore les lignes vides ;
• lignes 46-60 : le traitement des contribuables. Chaque ligne est validée par l'expression régulière de la ligne 5 : un [o] ou un [n], puis deux entiers positifs, séparés par des espaces ([\s+]). Les parenthèses définissent trois groupes, que [group(1)], [group(2)] et [group(3)] rendent lorsque [matches()] a vérifié que toute la ligne correspond à l'expression. Une ligne incorrecte ([x] au lieu de [o/n], salaire négatif) est signalée comme telle ;
• lignes 62-77 : le calcul de l'impôt, en [BigDecimal]. La division du revenu par le nombre de parts peut ne pas tomber juste : on la fait avec une précision de 16 chiffres significatifs ([MathContext.DECIMAL64]). La première tranche dont la limite est supérieure ou égale au quotient familial est cherchée par un flux ; [orElse(tranches.getLast())] donne la dernière tranche si aucune ne convient. [intValue()] tronque l'impôt à l'euro ;
• les résultats donnent les valeurs de contrôle du fil rouge : 4 282 € pour un couple marié avec 2 enfants et 60 000 € de salaire, 2 959 € avec 3 enfants, 11 519 € pour une personne seule sans enfant.

Le calcul lui-même est synchrone : il ne fait aucune E/S et dure quelques microsecondes. Il serait inutile, et même nuisible, de le confier à un autre thread. On ne rend asynchrone que ce qui attend.

9.15. Les règles d'or

Résumons les règles de la programmation asynchrone avec Java :

• pour attendre, ne pas bloquer de thread du système : une attente (E/S, délai, autre opération) se fait soit dans un thread virtuel, soit en attachant une continuation à un [CompletableFuture] ;
• pour du code nouveau, préférer les threads virtuels : on écrit du code bloquant, séquentiel et lisible (boucles, [try / catch], variables locales ordinaires), exécuté par [Executors.newVirtualThreadPerTaskExecutor()] ; un thread virtuel par tâche, jamais de pool de threads virtuels ;
• [CompletableFuture] pour composer des opérations, ou pour utiliser une API qui rend des [CompletableFuture] ([HttpClient.sendAsync], bibliothèques réactives) ; on choisit alors l'exécuteur des continuations ([xxxAsync(..., executeur)]) et on n'exécute jamais d'opération bloquante dans le pool commun ;
• pas de [join()] ni de [get()] dans une continuation ou dans le thread d'une interface graphique : ces attentes bloquent le thread qui les exécute. Leur place est dans le programme principal ou dans un thread virtuel ;
• les threads virtuels servent les attentes, pas les calculs : pour un calcul, on utilise un pool de threads de plate-forme dimensionné sur le nombre de processeurs, ou les flux parallèles ;
• limiter l'accès aux ressources rares avec un [Semaphore], et non par la taille d'un pool de threads ;
• l'annulation est coopérative : on annule une tâche par [Future.cancel(true)], qui interrompt son thread ; une tâche interrompue s'arrête proprement ; on ne masque jamais une [InterruptedException] (on la propage, ou on rétablit l'indicateur d'interruption). Et l'on se souvient que [CompletableFuture.cancel] n'interrompt rien ;
• limiter la durée des attentes : [orTimeout], [completeOnTimeout], [get(délai, unité)] ; et arrêter l'opération elle-même si l'attente est abandonnée ;
• lancer d'abord, attendre ensuite : pour mener plusieurs opérations indépendantes de front, on les lance toutes puis on les attend ([allOf], [invokeAll], la méthode [close] de l'exécuteur) ;
• une opération lancée doit être attendue ou observée : une opération oubliée cache ses exceptions ([exceptionally], [whenComplete] pour les traiter) ; en remontant une [CompletionException] ou une [ExecutionException], on cherche la cause d'origine.

Le tableau suivant résume les équivalences entre le monde des threads de plate-forme et des attentes bloquantes (chapitre précédent), celui des [CompletableFuture] et celui des threads virtuels :

Threads de plate-forme, attentes bloquantes	CompletableFuture	Threads virtuels
[Thread.sleep(1000)] (bloque le thread)	[new CompletableFuture<Void>().completeOnTimeout(null, 1, SECONDS)]	[Thread.sleep(1000)] (libère le porteur)
[thread.start()] puis [thread.join()]	[supplyAsync(...)] puis [join()] ou une continuation	[executeur.submit(...)] puis [get()]
attendre plusieurs threads ([join] en boucle)	[allOf(...)], [anyOf(...)]	[invokeAll(...)], [ExecutorCompletionService], [close()]
méthodes de rappel ([CompletionHandler])	[thenApply], [thenCompose], [thenCombine]	le code qui suit l'appel bloquant
[try / catch] autour de [get()] ([ExecutionException])	[exceptionally], [handle], [whenComplete]	[try / catch] ordinaire
[thread.interrupt()]	[cancel(...)] (n'interrompt rien)	[future.cancel(true)] (interrompt le thread)
[future.get(délai, unité)]	[orTimeout(...)], [completeOnTimeout(...)]	[future.get(délai, unité)]
pool de threads de taille fixe	exécuteur des continuations [xxxAsync]	[Semaphore]
[BlockingQueue] (bloque le thread)	[Flow], [SubmissionPublisher]	[BlockingQueue] (libère le porteur)
[ScheduledExecutorService]	[completeOnTimeout], [delayedExecutor]	[Thread.sleep] dans une boucle

Les chapitres suivants appliqueront ces règles : accès aux bases de données avec JDBC (chapitre 10) et Hibernate (chapitre 11), dont les API sont bloquantes et se marient naturellement avec les threads virtuels, serveur TCP qui sert chaque client par un thread virtuel et client [HttpClient] (chapitre 12), services web Spring Boot, dont le serveur Tomcat peut traiter chaque requête par un thread virtuel, et leurs clients (chapitre 13).