CompletableFuture : exécution asynchrone et combinaison de résultats

1. Vue d’ensemble

Une API doit souvent appeler plusieurs services : récupérer les informations client, les commandes, les frais d’expédition. Si vous les appelez séquentiellement, le temps total égale la somme des latences de chaque appel. CompletableFuture vous permet d’exécuter les appels indépendants en parallèle, puis de combiner les résultats lorsque tout est terminé.

Après avoir lu cet article, vous saurez comment créer des tâches asynchrones, chaîner les étapes de traitement (thenApply, thenCompose) et combiner les résultats (thenCombine, allOf, anyOf). Vous saurez aussi comment gérer les erreurs, définir des timeouts et écrire des tests pour le code asynchrone.

2. Préparation de l’exemple

Nous utilisons trois services simulés. Chaque service « dort » pendant un certain temps pour simuler la latence réseau :

import java.math.BigDecimal;
import java.util.List;

record Customer(long id, String name, String city) {}
record Order(long id, long customerId, BigDecimal total) {}

final class Delay {
    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }
}

class CustomerService {
    Customer findById(long id) {
        Delay.sleep(300);
        return new Customer(id, "Lan", "Hue");
    }
}

class OrderService {
    List<Order> findByCustomer(long customerId) {
        Delay.sleep(400);
        return List.of(
            new Order(1, customerId, new BigDecimal("150000")),
            new Order(2, customerId, new BigDecimal("90000")));
    }
}

class ShippingService {
    BigDecimal feeFor(String city) {
        Delay.sleep(200);
        return "Hue".equals(city) ? new BigDecimal("30000") : new BigDecimal("45000");
    }
}

Ces trois services n’utilisent que le JDK. Vous n’avez besoin d’ajouter aucune bibliothèque pour les exécuter.

3. Créer des tâches asynchrones

3.1. supplyAsync et runAsync

supplyAsync exécute un Supplier et retourne CompletableFuture<T>. runAsync exécute un Runnable sans résultat et retourne CompletableFuture<Void>:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CreateDemo {
    public static void main(String[] args) {
        var customerService = new CustomerService();

        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            CompletableFuture<Customer> customerFuture =
                CompletableFuture.supplyAsync(() -> customerService.findById(42L), executor);

            CompletableFuture<Void> auditFuture =
                CompletableFuture.runAsync(() -> System.out.println("Enregistrer l'accès"), executor);

            System.out.println("Le thread principal n'est pas bloqué");
            System.out.println(customerFuture.join());
            auditFuture.join();
        }
    }
}

Le thread principal n'est pas bloqué
Enregistrer l'accès
Customer[id=42, name=Lan, city=Hue]

L’ordre des deux premières lignes peut être inversé, car elles s’exécutent sur deux threads différents. Vous devez faire attention à quelques points :

• Le deuxième paramètre est un Executor. S’il est omis, la tâche s’exécute sur ForkJoinPool.commonPool().
• Executors.newVirtualThreadPerTaskExecutor() (Java 21) crée un virtual thread pour chaque tâche. Ce choix est approprié pour les tâches I/O comme les appels HTTP ou les requêtes de base de données.
• ExecutorService est AutoCloseable depuis Java 19. Le bloc try-with-resources attend que les tâches se terminent avant de fermer l’executor.
• join() bloque le thread actuel jusqu’à ce qu’il y ait un résultat. Si la tâche échoue, join() lève CompletionException (unchecked). En revanche, get() lève ExecutionException (checked).

3.2. Compléter manuellement un future

Parfois, le résultat provient d’un callback plutôt que d’une lambda. Dans ce cas, vous créez un future vide et le complétez vous-même :

CompletableFuture<String> paymentResult = new CompletableFuture<>();

// Quelque part, par exemple dans le callback d'une passerelle de paiement :
paymentResult.complete("PAID");

System.out.println(paymentResult.join()); // PAID

Pour signaler une erreur, vous appelez completeExceptionally(new RuntimeException("Carte refusée")). Quand vous disposez déjà d’une valeur, CompletableFuture.completedFuture(value) est la façon la plus concise de créer un future complété.

4. Chaîner les étapes de traitement

4.1. thenApply et thenAccept

thenApply transforme le résultat, similaire à map de Stream. thenAccept consomme le résultat et ne retourne rien :

CompletableFuture<String> greeting = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenApply(Customer::name)
    .thenApply(name -> "Bonjour, " + name);

greeting.thenAccept(System.out::println).join();

Bonjour, Lan

Chaque appel à thenApply retourne un nouveau CompletableFuture. Le future d’origine n’est pas modifié.

4.2. thenCompose quand l’étape suivante est aussi asynchrone

Pour récupérer les commandes, nous avons besoin du customerId de l’étape précédente. Si vous utilisez thenApply avec une fonction qui retourne un future, le résultat sera CompletableFuture<CompletableFuture<List<Order>>>. thenCompose « aplatit » ce type imbriqué, similaire à flatMap:

CompletableFuture<List<Order>> ordersFuture = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenCompose(customer -> CompletableFuture.supplyAsync(
        () -> orderService.findByCustomer(customer.id()), executor));

System.out.println(ordersFuture.join().size()); // 2

Ces deux étapes dépendent l’une de l’autre, elles s’exécutent donc séquentiellement. Le temps total est d’environ 700 ms. thenCompose ne rend pas les choses plus rapides. Il vous aide à chaîner les étapes sans bloquer aucun thread.

5. Combiner les résultats

C’est avec les tâches indépendantes que CompletableFuture est vraiment puissant.

5.1. thenCombine pour deux futures

Supposons que nous connaissions déjà la ville du client et que nous voulions récupérer simultanément la liste des commandes et les frais d’expédition :

long start = System.currentTimeMillis();

CompletableFuture<List<Order>> orders =
    CompletableFuture.supplyAsync(() -> orderService.findByCustomer(42L), executor);
CompletableFuture<BigDecimal> shippingFee =
    CompletableFuture.supplyAsync(() -> shippingService.feeFor("Hue"), executor);

CompletableFuture<BigDecimal> grandTotal = orders.thenCombine(shippingFee,
    (orderList, fee) -> orderList.stream()
        .map(Order::total)
        .reduce(BigDecimal.ZERO, BigDecimal::add)
        .add(fee));

System.out.println(grandTotal.join());
System.out.println("Temps écoulé: " + (System.currentTimeMillis() - start) + " ms");

270000
Temps écoulé: 405 ms

Les deux appels s’exécutent en parallèle. Le temps total est approximativement celui de l’appel le plus lent (400 ms), au lieu de 600 ms si les appels étaient séquentiels. La fonction du deuxième paramètre de thenCombine s’exécute uniquement lorsque les deux futures sont complétés.

5.2. allOf pour plusieurs futures

Quand vous avez plus de deux tâches, utilisez allOf. Il retourne CompletableFuture<Void>, donc vous devez récupérer vous-même le résultat de chaque future enfant :

List<String> cities = List.of("Hue", "Hanoi", "Danang");

List<CompletableFuture<BigDecimal>> fees = cities.stream()
    .map(city -> CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor))
    .toList();

CompletableFuture<List<BigDecimal>> allFees = CompletableFuture
    .allOf(fees.toArray(CompletableFuture[]::new))
    .thenApply(ignored -> fees.stream()
        .map(CompletableFuture::join)
        .toList());

System.out.println(allFees.join());

[30000, 45000, 45000]

À l’intérieur de thenApply, tous les futures enfants sont complétés. Par conséquent, join() retourne immédiatement sans bloquer. L’ordre des résultats correspond à l’ordre dans cities.

5.3. anyOf pour obtenir le résultat le plus rapide

anyOf se complète dès qu’un des futures enfants est complété. C’est utile quand vous appelez plusieurs répliques du même service :

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> { Delay.sleep(500); return "replica-A"; }, executor),
    CompletableFuture.supplyAsync(() -> { Delay.sleep(100); return "replica-B"; }, executor));

System.out.println(fastest.join()); // replica-B

Le type de retour est CompletableFuture<Object>, donc vous devez faire un cast vous-même. Notez également que si le future le plus rapide se termine par une erreur, anyOf se termine aussi par cette erreur.

6. Gérer les erreurs et les timeouts

6.1. exceptionally et handle

Une exception à n’importe quelle étape se propage aux étapes suivantes. Ces étapes sont ignorées jusqu’à ce qu’une étape de gestion d’erreur soit rencontrée :

CompletableFuture<BigDecimal> fee = CompletableFuture
    .supplyAsync(() -> {
        throw new IllegalStateException("Service d'expédition indisponible");
    }, executor)
    .thenApply(value -> (BigDecimal) value)          // ignorée
    .exceptionally(ex -> {
        System.out.println("Erreur: " + ex.getMessage());
        return new BigDecimal("50000");              // valeur de secours
    });

System.out.println(fee.join());

Erreur: java.lang.IllegalStateException: Service d'expédition indisponible
50000

Le paramètre ex est généralement une CompletionException qui enveloppe l’exception originale. C’est pourquoi le message a un préfixe du nom de la classe. Vous utilisez ex.getCause() pour obtenir l’erreur réelle.

handle reçoit à la fois le résultat et l’erreur. Il est toujours appelé, que la tâche réussisse ou échoue :

CompletableFuture<String> status = CompletableFuture
    .supplyAsync(() -> orderService.findByCustomer(42L), executor)
    .handle((orderList, ex) -> ex == null
        ? "Il y a " + orderList.size() + " commandes"
        : "Impossible de charger les commandes");

Il y a aussi whenComplete. Il reçoit à la fois le résultat et l’erreur, mais ne modifie pas le résultat. Cette méthode est appropriée pour enregistrer des événements.

6.2. orTimeout et completeOnTimeout

Vous ne devriez jamais attendre indéfiniment un service externe. Java 9+ fournit deux méthodes pour cela :

import java.util.concurrent.TimeUnit;

CompletableFuture<BigDecimal> withDefault = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .completeOnTimeout(new BigDecimal("45000"), 500, TimeUnit.MILLISECONDS);

System.out.println(withDefault.join()); // 45000 après ~500 ms

CompletableFuture<BigDecimal> strict = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .orTimeout(500, TimeUnit.MILLISECONDS);
// strict.join() lève CompletionException avec TimeoutException comme cause

completeOnTimeout retourne une valeur par défaut en cas de timeout. orTimeout termine le future avec une TimeoutException.

Note: Le timeout complète seulement le future. Il n’arrête pas la tâche sous-jacente. Le thread continue à « dormir » pendant 2 secondes. cancel(true) ne peut pas non plus interrompre le thread en cours d’exécution. Si vous avez besoin d’annuler vraiment, vous devez vérifier un drapeau d’annulation dans la tâche vous-même ou utiliser le timeout du client HTTP lui-même.

7. Écrire des tests

Si vous utilisez Spring Boot, spring-boot-starter-test inclut déjà JUnit 5 et AssertJ. Pour un projet Java pur, ajoutez :

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.assertj</groupId>
    <artifactId>assertj-core</artifactId>
    <version>3.26.3</version>
    <scope>test</scope>
</dependency>

D’abord, encapsulez la logique dans une classe pour faciliter les tests :

class CheckoutService {
    private final OrderService orderService;
    private final ShippingService shippingService;
    private final Executor executor;

    CheckoutService(OrderService o, ShippingService s, Executor executor) {
        this.orderService = o;
        this.shippingService = s;
        this.executor = executor;
    }

    CompletableFuture<BigDecimal> grandTotal(long customerId, String city) {
        var orders = CompletableFuture.supplyAsync(() -> orderService.findByCustomer(customerId), executor);
        var fee = CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor);
        return orders.thenCombine(fee, (list, f) -> list.stream()
            .map(Order::total)
            .reduce(BigDecimal.ZERO, BigDecimal::add)
            .add(f));
    }
}

Ensuite, écrivez des tests pour les cas de succès et d’échec :

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import org.junit.jupiter.api.Test;

class CheckoutServiceTest {

    @Test
    void shouldCombineOrdersAndShippingFee() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), new ShippingService(), executor);

            assertThat(service.grandTotal(42L, "Hue"))
                .succeedsWithin(Duration.ofSeconds(2))
                .isEqualTo(new BigDecimal("270000"));
        }
    }

    @Test
    void shouldPropagateShippingFailure() {
        ShippingService broken = new ShippingService() {
            @Override
            BigDecimal feeFor(String city) {
                throw new IllegalStateException("indisponible");
            }
        };

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), broken, executor);

            assertThatThrownBy(() -> service.grandTotal(42L, "Hue").join())
                .isInstanceOf(CompletionException.class)
                .hasCauseInstanceOf(IllegalStateException.class);
        }
    }
}

succeedsWithin attend au maximum 2 secondes puis vérifie la valeur. Ainsi, le test ne s’éternise pas si le code a un bug. Dans le deuxième test, l’erreur du ShippingService se propage à travers thenCombine et est enveloppée dans une CompletionException.

Dans les tests, vous pouvez passer Runnable::run comme Executor pour que tout s’exécute de manière synchrone sur le thread du test. Cela rend les tests plus rapides et les résultats plus stables.

8. Erreurs courantes

• Exécuter I/O sur le commonPool. Ce pool a un nombre de threads égal au nombre de CPU moins un, et il est partagé par toute l’application (y compris les parallel streams). Quelques appels HTTP lents peuvent bloquer le pool entier. Pour I/O, vous devriez passer un executor dédié, de préférence des virtual threads.
• Appeler join() trop tôt. Écrire supplyAsync(a).join() puis supplyAsync(b).join() reste séquentiel. Vous devez lancer tous les futures d’abord, puis les combiner ou les attendre.
• Avaler les exceptions. Si personne n’appelle join(), get() ou n’attache pas exceptionally/whenComplete, l’erreur disparaîtra sans laisser de trace.
• Confondre thenApply avec thenApplyAsync. thenApply peut s’exécuter sur le thread qui vient de compléter l’étape précédente, ou sur le thread appelant si le future est déjà terminé. Si l’étape suivante est lourde ou bloquante, utilisez la variante ...Async(fn, executor).
• Ne pas définir de timeout. Un service qui traîne bloquera votre requête en conséquence.
• Le programme se termine trop tôt. Les threads du commonPool sont des daemon threads. Si main se termine avant que la tâche ne soit terminée, le résultat ne sera jamais affiché.

Note: Vous ne devriez pas utiliser CompletableFuture s’il n’y a qu’un seul appel et que vous attendez immédiatement le résultat. À partir de Java 21, avec des virtual threads, du code synchrone ordinaire est généralement plus lisible sans perte de performance. CompletableFuture ne vaut vraiment la peine d’être utilisé que si vous avez besoin d’exécuter plusieurs tâches indépendantes en parallèle, ou si vous avez besoin d’un pipeline non-bloquant.

9. Conclusion

CompletableFuture vous aide à exécuter les tâches indépendantes en parallèle et à combiner les résultats sans avoir à gérer les threads vous-même. Un résumé rapide :

• Créer une tâche: supplyAsync/runAsync, toujours passer un executor approprié.
• Chaîner les étapes: thenApply pour transformer une valeur, thenCompose quand l’étape suivante retourne un future.
• Combiner: thenCombine pour deux futures, allOf pour plusieurs futures, anyOf pour obtenir le résultat le plus rapide.
• Erreurs et timeouts: exceptionally, handle, orTimeout, completeOnTimeout.

Dans les tests, utilisez succeedsWithin d’AssertJ pour vérifier le résultat sans craindre que le test se bloque indéfiniment.