Asynchronous Programming in Java

Table of Contents

1. Course Overview
2. Asynchronous data access on the web

• Introduction and agenda
• Java version required
• Why asynchronous code improves throughput
• Prerequisites
• Course Agenda
• Synchronous code vs asynchronous code
• Non-concurrent asynchronous code
• Avoid blocking of long tasks
• Use concurrency to write asynchronous code
• Start a task with an ExecutorService
• Demo: synchronous launch of multiple tasks
• Demo: improve throughput with an ExecutorService
• Demo: use the CompletionStage API
• Module 2 Summary

3. Trigger a task on the result of another task

• Introduction and agenda
• Avoid application thread blocking
• CompletionStage and CompletableFuture — details
• Send task result to a function asynchronously
• Design an efficient asynchronous processing pipeline
• Demo: first asynchronous processing pipeline
• Module 3 Summary

4. Split a result into several asynchronous tasks

• Introduction and agenda
• Get a result faster with anyOf
• Launch several tasks in parallel with allOf
• Parse the CompletableFuture returned by allOf
• Compose long tasks to avoid blocking (thenCompose)
• Demo: find the best element of a collection with allOf
• Demo: get a result faster with anyOf
• Demo: compose quotation and weather retrieval
• Module 4 Summary

5. Control which thread executes a task

• Introduction and agenda
• The default threads used by the CompletionStage API
• Choose an Executor for your asynchronous tasks
• Demo: specifying Executors for tasks
• Module 5 Summary

6. Report and recover from errors

• Introduction and agenda
• What happens when a task fails with an exception?
• Log and recover with exceptionally
• Handle exceptions with whenComplete and handle
• Demo: handling exceptions in the TravelPage application
• Module 6 Summary

7. Closing Remarks

• Introduction and agenda
• Patterns for creating an asynchronous task
• Patterns for modeling tasks
• Patterns for chaining and composing tasks
• Patterns for handling exceptions
• Patterns for better performance
• Final Course Summary

8. Demo source code

• Module 2 — A: Accessing Data Asynchronously
• Module 3 — B: Triggering a Task
• Module 4 — C: Splitting a Result
• Module 5 — D: Controlling Threads
• Module 6 — E: Reporting Errors

1. Course Overview

This course teaches how to create efficient and secure asynchronous data processing pipelines using the JDK’s CompletableFuture API. In approximately two hours, you will learn to:

• Create simple asynchronous tasks.
• Chain these tasks and trigger one task on the result of another.
• Understand which thread the API chooses to execute your tasks.
• Specify Executors to control threads.
• All this illustrated on real examples that you can run yourself.

By the end of this course, you will be fully equipped to understand the structure of the CompletableFuture API and navigate its many methods.

2. Asynchronous access to data on the web

2.1 Introduction and agenda

This course focuses on understanding how to write asynchronous code with the API provided by the JDK. Asynchronous programming is a very popular topic, and there are many tools, APIs and frameworks outside of the JDK for creating asynchronous applications. This course covers the CompletionStage and CompletableFuture APIs, which are part of the JDK.

If you understand the asynchronous principles covered here, you will find it easier to use any other API because they are all based on the same ideas and principles.

2.2 Java version required

This is a Java SE course. The version used for the examples is Java SE 17, but all code works correctly with all versions of Java after Java SE 8, because the CompletionStage API was added in Java SE 8.

Important points:

• Some methods were added after Java 8 in the CompletionStage API, and the instructor mentions them when using them.
• No methods have been removed. This API, like the rest of Java, is 100% backwards compatible.
• The IDE used is IntelliJ IDEA Ultimate Edition (the free Community Edition is also suitable). Any Java IDE works: Eclipse, NetBeans, VS Code, etc.

2.3 Why asynchronous code improves throughput

Why do you want to write asynchronous code? Writing asynchronous code is complex, produces code difficult to test and debug, and probably difficult to read. This comes at a cost to yourself and your application in the long run.

Writing asynchronous code is above all improving the throughput of your application:

• Web application → throughput = number of HTTP requests processed per second/minute.
• Messaging application → throughput = number of messages processed per second/minute.

It’s all about network communication, disk communication, databases, and throughput is how much information you can process per second or per minute.

2.4 Prerequisites

To take this course, you need:

• Basic Java knowledge: write a simple program with a main method.
• Framework collections and lambda expressions: used extensively throughout the course.
• Basics of concurrent programming: know what a thread is, have used ExecutorService, Runnable, Callable, Future.

2.5 Course Agenda

Course structure:

2.6 Synchronous code vs asynchronous code

Synchronous code: To run task B, you must wait for task A to complete.

• Sometimes logical: if B needs the result of A.
• But: During this time, your system remains inactive. It could execute a task C independent of A and B.
• If B doesn’t need information from A, waiting is wasteful.
• It may take several hundreds of milliseconds for an HTTP response to be available → your synchronous code does nothing during this time.

Asynchronous code: does not wait **. It launches tasks that will be executed at a future time.

• When exactly? We don’t know.
• How many times? 0, 1, or N times — it depends on the context.
• Example: the code of a lambda passed to forEach is asynchronous code. It will be executed an unspecified number of times upon declaration.

Asynchronous programming is not necessarily linked to concurrent programming. A forEach is not concurrent, but it is asynchronous. The CompletionStage API builds asynchrony over concurrency, but it is not required.

2.7 Non-concurrent asynchronous code

// Exemple de code asynchrone NON concurrent
List<String> strings = List.of("a", "b", "c");
strings.forEach(s -> System.out.println(s));
// Le System.out.println est du code asynchrone : il sera exécuté
// à un moment futur, un nombre de fois indéterminé (0, 1 ou N).

This code illustrates that asynchrony does not require concurrency.

2.8 Avoid blocking long-running tasks

Blocking code vs non-blocking code:

• The synchronous code is always blocking.
• The real question: how long will this block your processing pipeline?

If you write an application that blocks on every input/output operation, you will have difficulty maintaining high throughput.

Examples:

• Block on an HTTP request for several hundred milliseconds → major impact on throughput.
• Block in a forEach which takes little time to process → negligible impact.

What you will learn in this course: write asynchronous code to avoid blocking your application for several hundred milliseconds. The first step is to identify the portions of code in your application that are blocking for this duration, using the CompletionStage API — asynchronous programming built on concurrency (multithreaded applications).

2.9 Using concurrency to write asynchronous code

Concurrency in this context means running a specific task (querying the web) in a different thread than the one running your application.

• Executing code asynchronously in another thread helps avoid blocking your main thread.
• Your main thread may start other tasks while the first task is blocking in another thread.
• You need a way to know if a task started in another thread completed normally (with a result) or exceptionally (with an exception).
• If you need the result, you should transfer it from the other thread to your main thread.

2.10 Launch a task with an ExecutorService

The first pattern available in the JDK is the ExecutorService pattern (since Java SE 5, package java.util.concurrent):

1. Create an ExecutorService (a thread pool with a queue).
2. Submit a task (Runnable or Callable) to this ExecutorService.
3. The task is added to the queue, and the first available thread picks it up and executes it.
4. You immediately get a Future object in return.

Future (interface since Java 5):

• Models the object returned by the ExecutorService.
• Allows you to know if the task is completed or not.
• If completed, gives the result or exception.

Limitation: getting the result with future.get() is a synchronous blocking call. We are therefore not yet truly asynchronous.

2.11 Demo: Synchronous launch of multiple tasks

This first example establishes the synchronous benchmark for measuring the performance gain of subsequent asynchronous approaches.

Each Callable<Quotation> simulates a network request with Thread.sleep(random.nextInt(80, 120)) (80 to 120ms). Sequential execution takes ~300ms (3 × ~100ms).

package org.paumard.async;

import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;

public class A_RunSynchronousTasks {

    record Quotation(String server, int amount) {
    }

    public static void main(String[] args) {
        run();
    }

    public static void run() {
        Random random = new Random();

        Callable<Quotation> fetchQuotationA =
              () -> {
                  Thread.sleep(random.nextInt(80, 120));
                  return new Quotation("Server A", random.nextInt(40, 60));
              };
        Callable<Quotation> fetchQuotationB =
              () -> {
                  Thread.sleep(random.nextInt(80, 120));
                  return new Quotation("Server B", random.nextInt(30, 70));
              };
        Callable<Quotation> fetchQuotationC =
              () -> {
                  Thread.sleep(random.nextInt(80, 120));
                  return new Quotation("Server C", random.nextInt(40, 80));
              };

        var quotationTasks =
              List.of(fetchQuotationA, fetchQuotationB, fetchQuotationC);

        Instant begin = Instant.now();
        Quotation bestQuotation =
        quotationTasks.stream()
              .map(A_RunSynchronousTasks::fetchQuotation)
              .min(Comparator.comparing(Quotation::amount))
              .orElseThrow();
        Instant end = Instant.now();
        Duration duration = Duration.between(begin, end);
        System.out.println("Best quotation [SYNC ] = " + bestQuotation +
              " (" + duration.toMillis() + "ms)");
    }

    private static Quotation fetchQuotation(Callable<Quotation> task) {
        try {
            return task.call();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

Important points:

• record Quotation(String server, int amount) — uses Java 16+ records.
• Thread.sleep(random.nextInt(80, 120)) — simulates a network request (80 to 120ms). The nextInt(min, max) syntax is available since Java 17.
• The orElseThrow() method is preferable to get() because it explicitly reminds that get() can throw an exception on an empty Optional.
• Typical execution time: ~300ms (sequential execution of the 3 tasks).

2.12 Demo: improve throughput with an ExecutorService

Executing the same Callable in parallel with a 4-thread ExecutorService.

package org.paumard.async;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class B_RunExecutorTasks {

    record Quotation(String server, int amount) {
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        run();
    }

    public static void run() throws ExecutionException, InterruptedException {

        Random random = new Random();

        Callable<Quotation> fetchQuotationA =
              () -> {
                  Thread.sleep(random.nextInt(80, 120));
                  return new Quotation("Server A", random.nextInt(40, 60));
              };
        Callable<Quotation> fetchQuotationB =
              () -> {
                  Thread.sleep(random.nextInt(80, 120));
                  return new Quotation("Server B", random.nextInt(30, 70));
              };
        Callable<Quotation> fetchQuotationC =
              () -> {
                  Thread.sleep(random.nextInt(80, 120));
                  return new Quotation("Server C", random.nextInt(40, 80));
              };

        var quotationTasks =
              List.of(fetchQuotationA, fetchQuotationB, fetchQuotationC);

        var executor = Executors.newFixedThreadPool(4);
        Instant begin = Instant.now();

        List<Future<Quotation>> futures = new ArrayList<>();
        for (Callable<Quotation> task : quotationTasks) {
            Future<Quotation> future = executor.submit(task);
            futures.add(future);
        }

        List<Quotation> quotations = new ArrayList<>();
        for (Future<Quotation> future : futures) {
            Quotation quotation = future.get();
            quotations.add(quotation);
        }

        Quotation bestQuotation =
        quotations.stream()
              .min(Comparator.comparing(Quotation::amount))
              .orElseThrow();

        Instant end = Instant.now();
        Duration duration = Duration.between(begin, end);
        System.out.println("Best quotation [ES   ] = " + bestQuotation +
              " (" + duration.toMillis() + "ms)");

        executor.shutdown();
    }
}

Important points:

• Executors.newFixedThreadPool(4) — 4 threads for 3 tasks (1 spare thread).
• Tasks are submitted one by one via executor.submit(task), each returns a Future<Quotation>.
• The future.get() call is always blocking — this is the major problem with this pattern: we bring the result back to the main thread, which blocks the latter.
• Execution time: ~120ms (the 3 tasks run in parallel, we wait for the slowest one).
• Don’t forget executor.shutdown() to free resources.

2.13 Demo: using the CompletionStage API

Replacing Callable with Supplier and using CompletableFuture.supplyAsync().

package org.paumard.async;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.function.Supplier;

public class C_RunAsyncTasks {

    record Quotation(String server, int amount) {
    }

    public static void main(String[] args) {
        run();
    }

    public static void run() {

        Random random = new Random();

        Supplier<Quotation> fetchQuotationA =
              () -> {
                  try {
                      Thread.sleep(random.nextInt(80, 120));
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
                  return new Quotation("Server A", random.nextInt(40, 60));
              };
        Supplier<Quotation> fetchQuotationB =
              () -> {
                  try {
                      Thread.sleep(random.nextInt(80, 120));
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
                  return new Quotation("Server B", random.nextInt(30, 70));
              };
        Supplier<Quotation> fetchQuotationC =
              () -> {
                  try {
                      Thread.sleep(random.nextInt(80, 120));
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
                  return new Quotation("Server C", random.nextInt(40, 80));
              };

        var quotationTasks =
              List.of(fetchQuotationA, fetchQuotationB, fetchQuotationC);

        Instant begin = Instant.now();

        List<CompletableFuture<Quotation>> futures = new ArrayList<>();
        for (Supplier<Quotation> task : quotationTasks) {
            CompletableFuture<Quotation> future =
                  CompletableFuture.supplyAsync(task);
            futures.add(future);
        }

        List<Quotation> quotations = new ArrayList<>();
        for (CompletableFuture<Quotation> future : futures) {
            Quotation quotation = future.join();
            quotations.add(quotation);
        }

        Quotation bestQuotation =
        quotations.stream()
              .min(Comparator.comparing(Quotation::amount))
              .orElseThrow();

        Instant end = Instant.now();
        Duration duration = Duration.between(begin, end);
        System.out.println("Best quotation [ASYNC] = " + bestQuotation +
              " (" + duration.toMillis() + "ms)");
    }
}

Key differences from ExecutorService:

Important note: The call to future.join() is still blocking here. This is what the next module will solve with thenApply, thenAccept, etc.

2.14 Module 2 Summary

This module taught you:

1. The difference between synchronous and asynchronous code, and how avoiding main thread blocking can speed up your application’s throughput.
2. How to move the execution of blocking tasks to another thread so that they execute in parallel without blocking the main thread.
3. What is needed now: a way to trigger a task on the result of another task without returning to the main thread — this is the subject of the next module.

3. Trigger a task on the result of another task

3.1 Introduction and agenda

Now that you know how to start a task asynchronously with the CompletionStage API, you will learn the next step: how to chain tasks in a new way. At the end of this module, you will be able to create your first asynchronous processing pipeline — a first task and several downstream tasks automatically triggered on the result of the first.

3.2 Avoid blocking application threads

Example: Travel agency system

Imagine a travel agency system that must:

1. Query a server for quotes.
2. Send an email with the information from the first server.
3. Save information to the database.

The problem with the ExecutorService pattern (or with the elements of the CompletionStage API seen previously): the main thread is blocked waiting for information to return from the Future objects. The ExecutorService threads are not blocked, but your main thread is — and a blocked thread is costly.

If your thread is blocked, it cannot do anything else. If you want to do something else, you’ll probably have to start a new thread, which is also expensive.

What you need: a way to declare that, as soon as the quote is available, the result should be passed to another task that will process it at that time — without returning to the main thread. This is exactly what the CompletionStage API does for you.

3.3 CompletionStage and CompletableFuture — details

Some important details about the CompletionStage API:

• CompletionStage is the name of an interface — it is the center of this API.
• CompletableFuture is the class provided by the JDK that implements CompletionStage.
• CompletableFuture also implements the Future interface.
• So the object you are working with is both a CompletionStage and a Future.

The terms “CompletionStage API” and “CompletableFuture API” mean the same thing. Both names are used interchangeably in the industry.

CompletionStage (interface)
      ↑
CompletableFuture (classe)  ←── implémente aussi Future (interface)

3.4 Send the result of a task to a function asynchronously

Basic pattern:

// Lancer une tâche asynchrone
CompletableFuture<Quotation> quotationCF =
    CompletableFuture.supplyAsync(fetchQuotation);

// Chaîner une tâche sur le résultat — NON BLOQUANT
CompletableFuture<Void> emailCF =
    quotationCF.thenApply(quotation -> sendEmail(quotation))
               .thenApply(emailResult -> writeToDatabase(emailResult))
               .thenApply(dbResult   -> updateGUI(dbResult));

Available methods for chaining:

thenApply is the first pattern to chain tasks and create a fully asynchronous data processing pipeline.

This declaration is non-blocking. Your function will be executed at some future time, making it asynchronous code. This thenApply gives you another CompletableFuture that you can use to chain another task over and over again.

3.5 Design an efficient asynchronous processing pipeline

Several tips for designing an efficient asynchronous pipeline:

1. Never block your main thread. Blocking the main thread is bad for your throughput.

2. Divide processing into small tasks — each task does one thing: query a server, database, disk resource, etc.

3. Depending on the task type, use the corresponding method of the CompletableFuture API.

4. A single CompletableFuture can trigger as many tasks as necessary. You can call thenApply and the other methods as many times as you want. This is not a Stream API.

5. A single task can trigger several downstream tasks, and the reverse is also possible: you can trigger a task on the result of several tasks (methods exist for this).

3.6 Demo: First Asynchronous Processing Pipeline

Replaced blocking future.join() calls with non-blocking thenAccept calls.

package org.paumard.chaining;

import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.function.Supplier;

public class A_TriggerTasks {

    record Quotation(String server, int amount) {
    }

    public static void main(String[] args) throws InterruptedException {
        run();
    }

    public static void run() throws InterruptedException {

        Random random = new Random();

        Supplier<Quotation> fetchQuotationA =
              () -> {
                  try {
                      Thread.sleep(random.nextInt(80, 120));
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
                  return new Quotation("Server A", random.nextInt(40, 60));
              };
        Supplier<Quotation> fetchQuotationB =
              () -> {
                  try {
                      Thread.sleep(random.nextInt(80, 120));
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
                  return new Quotation("Server B", random.nextInt(30, 70));
              };
        Supplier<Quotation> fetchQuotationC =
              () -> {
                  try {
                      Thread.sleep(random.nextInt(80, 120));
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
                  return new Quotation("Server C", random.nextInt(40, 80));
              };

        var quotationTasks =
              List.of(fetchQuotationA, fetchQuotationB, fetchQuotationC);

        Instant begin = Instant.now();

        List<CompletableFuture<Quotation>> futures = new ArrayList<>();
        for (Supplier<Quotation> task : quotationTasks) {
            CompletableFuture<Quotation> future =
                  CompletableFuture.supplyAsync(task);
            futures.add(future);
        }

        // Utilisation de ConcurrentLinkedDeque car plusieurs threads vont y écrire
        Collection<Quotation> quotations = new ConcurrentLinkedDeque<>();
        List<CompletableFuture<Void>> voids = new ArrayList<>();
        for (CompletableFuture<Quotation> future : futures) {

            // Afficher chaque quotation dès qu'elle est disponible (non bloquant)
            future.thenAccept(System.out::println);

            // Ajouter à la collection thread-safe dès que le résultat est disponible
            CompletableFuture<Void> accept =
                  future.thenAccept(quotations::add);
            voids.add(accept);
        }

        // Attendre que tous les thenAccept soient terminés
        for (CompletableFuture<Void> v : voids) {
            v.join();
        }
        System.out.println("quotations = " + quotations);

        Quotation bestQuotation =
              quotations.stream()
                    .min(Comparator.comparing(Quotation::amount))
                    .orElseThrow();

        Instant end = Instant.now();
        Duration duration = Duration.between(begin, end);
        System.out.println("Best quotation [ASYNC] = " + bestQuotation +
              " (" + duration.toMillis() + "ms)");
    }
}

Key points from this demo:

• thenAccept: method of CompletableFuture which does not exist on Future. It takes a Consumer and is called in a non-blocking manner.
• Important behavior: If you do not call v.join() at the end, the program may terminate before the thenAccept callbacks have had time to execute. Indeed, the main thread finishes its instructions and the JVM stops, killing the ForkJoinPool daemon threads. This is why we must call join() on the CompletableFuture<Void> returned by thenAccept.
• ConcurrentLinkedDeque: thread-safe data structure used because multiple threads in the ForkJoinPool can write to it concurrently.
• A CompletableFuture can trigger several downstream tasks: here future.thenAccept(System.out::println) and future.thenAccept(quotations::add) are two independent chains on the same CompletableFuture.

3.7 Module 3 Summary

This module taught you:

1. How to chain tasks to create your first asynchronous processing pipelines.
2. That a task can trigger as many downstream tasks as necessary.
3. The limit: the chained tasks were in-memory tasks (synchronous). The next question: how to combine the result of a first asynchronous task with the result of a second also asynchronous task? This is what you will learn in the next module — the composition of CompletableFutures.

4. Divide a result into several asynchronous tasks

4.1 Introduction and agenda

You now know how to create tasks, launch them asynchronously and chain them. This module teaches you two elements of the API that give you everything you need to create any type of pipeline:

1. anyOf — launch several tasks at once, keep only the first result (eg: query several DNS servers, all results are equivalent).
2. allOf — run multiple tasks at once and need all results.
3. thenCompose — the composition of CompletableFutures, one of the great features of this API.

4.2 Get results faster with anyOf

Use case: query multiple weather servers at the same time. All results are equivalent. We want the first available result.

CompletableFuture.anyOf(CompletableFuture<?>... cfs)

• Takes a variable number of CompletableFuture (vararg).
• Returns a CompletableFuture<Object> which completes as soon as one of its arguments completes.
• It’s not a guarantee that it will be the fastest, but at least the result will be available as soon as a first result arrives.
• This method can also propagate exceptions.

// Pattern anyOf
CompletableFuture<Weather> taskA = CompletableFuture.supplyAsync(fetchWeatherA);
CompletableFuture<Weather> taskB = CompletableFuture.supplyAsync(fetchWeatherB);
CompletableFuture<Weather> taskC = CompletableFuture.supplyAsync(fetchWeatherC);

CompletableFuture<Object> anyOf = CompletableFuture.anyOf(taskA, taskB, taskC);

// anyOf retourne CompletableFuture<Object>, cast nécessaire
anyOf.thenApply(o -> (Weather) o)
     .thenAccept(System.out::println)
     .join();

4.3 Launch several tasks in parallel with allOf

Use case: query multiple quote servers in parallel, keep best price.

CompletableFuture.allOf(CompletableFuture<?>... cfs)

• Takes a variable number of CompletableFuture (vararg).
• Returns a CompletableFuture<Void> which completes when all the tasks passed as arguments are completed.
• Important: CompletableFuture<Void> carries no results. It only indicates that all your tasks are completed.
• allOf takes CompletableFutures of any type.

CompletableFuture<Void> allOf =
    CompletableFuture.allOf(
        quotationCF_A,
        quotationCF_B,
        quotationCF_C
    );

4.4 Parse the CompletableFuture returned by allOf

Since allOf returns a CompletableFuture<Void>, you have to use a trick with the Stream API to extract the results:

CompletableFuture<Void> allOf =
    CompletableFuture.allOf(quotationCFs.toArray(CompletableFuture[]::new));

Quotation bestQuotation =
    allOf.thenApply(
        v ->  // v est toujours null !
            quotationCFs.stream()
                .map(CompletableFuture::join)  // join() est non-bloquant ICI
                .min(Comparator.comparing(Quotation::amount))
                .orElseThrow()
    ).join();

Why join() is acceptable here:

Normally, join() is a blocking call to avoid. But here, the function passed to thenApply is called only when allOf completes, that is, when all the CompletableFutures have already produced their result. Calling join() in this context will not block the processing thread.

4.5 Compose long tasks to avoid blocking (thenCompose)

Problem: you need to create a TravelPage combining a quote (Quotation) and a weather forecast (Weather), each coming from an independent asynchronous task.

Bad approaches:

// ❌ MAUVAIS : appel bloquant dans le thread principal
TravelPage page = new TravelPage(bestQuotationCF.join(), weatherCF.join());

// ❌ MAUVAIS : appel bloquant dans une function thenApply
bestQuotationCF.thenApply(
    quotation -> new TravelPage(quotation, weatherCF.join()) // join() bloquant
);

Good approach — thenCompose:

// ✅ BON : composition non-bloquante
CompletableFuture<TravelPage> travelPageCF =
    bestQuotationCF.thenCompose(
        quotation -> weatherCF.thenApply(
            weather -> new TravelPage(quotation, weather)
        )
    );
travelPageCF.thenAccept(System.out::println).join();

thenCompose(Function<T, CompletableFuture<U>>)

• Takes a Function which itself returns a CompletableFuture.
• This is the equivalent of flatMap in the Stream API.
• This pattern is the one to use when you have two asynchronous tasks and you want to combine their results without blocking.

Alternative — thenCombine:

// thenCombine : déclenché quand les DEUX CompletableFutures sont complétés
CompletableFuture<TravelPage> pageCompletableFuture =
    bestQuotationCF.thenCombine(
        weatherCF,
        TravelPage::new  // BiFunction<Quotation, Weather, TravelPage>
    );

4.6 Demo: find the best element in a collection with allOf

package org.paumard.several;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class A_ReadingSeveralTasks {

    record Quotation(String server, int amount) {
    }

    public static void main(String[] args) throws InterruptedException {
        run();
    }

    public static void run() throws InterruptedException {

        Random random = new Random();

        List<Supplier<Quotation>> quotationTasks = buildQuotationTasks(random);

        List<CompletableFuture<Quotation>> quotationCFS = new ArrayList<>();
        for (Supplier<Quotation> task : quotationTasks) {
            CompletableFuture<Quotation> future = CompletableFuture.supplyAsync(task);
            quotationCFS.add(future);
        }

        // allOf : signal que toutes les tâches sont terminées
        CompletableFuture<Void> allOf =
              CompletableFuture.allOf(quotationCFS.toArray(CompletableFuture[]::new));

        // Quand allOf est complété, extraire le meilleur devis
        Quotation bestQuotation =
        allOf.thenApply(
          v ->
              quotationCFS.stream()
                    .map(CompletableFuture::join) // non-bloquant ici
                    .min(Comparator.comparing(Quotation::amount))
                    .orElseThrow()
        ).join(); // ce join() attend la fin du pipeline
        System.out.println("bestQuotation = " + bestQuotation);
    }

    private static List<Supplier<Quotation>> buildQuotationTasks(Random random) {
        // ... même code que précédemment
        Supplier<Quotation> fetchQuotationA =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Quotation("Server A", random.nextInt(40, 60));
              };
        // ... fetchQuotationB, fetchQuotationC similaires
        return List.of(fetchQuotationA, /*fetchQuotationB, fetchQuotationC*/null, null);
    }
}

Pattern toArray(CompletableFuture[]::new) :

// Conversion List<CompletableFuture<T>> → CompletableFuture<?>[]
CompletableFuture<?>[] array = quotationCFS.toArray(CompletableFuture[]::new);
CompletableFuture<Void> allOf = CompletableFuture.allOf(array);

4.7 Demo: get results faster with anyOf

package org.paumard.several;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class B_ReadingOneOfSeveralTasks {

    record Weather(String server, String weather) {
    }

    public static void main(String[] args) throws InterruptedException {
        run();
    }

    public static void run() throws InterruptedException {

        Random random = new Random();

        List<Supplier<Weather>> weatherTasks = buildWeatherTasks(random);

        List<CompletableFuture<Weather>> futures = new ArrayList<>();
        for (Supplier<Weather> task : weatherTasks) {
            CompletableFuture<Weather> future = CompletableFuture.supplyAsync(task);
            futures.add(future);
        }

        // anyOf retourne CompletableFuture<Object>
        CompletableFuture<Object> future =
              CompletableFuture.anyOf(futures.toArray(CompletableFuture[]::new));

        // cast nécessaire car anyOf est générique Object
        future.thenAccept(System.out::println).join();
    }
    // ...
}

Observed behavior of anyOf:

package org.paumard.several;

import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class E_AnyOf {

    record Weather(String server, String weather) {
    }

    public static void main(String[] args) {

        Supplier<Weather> fetchWeatherA =
              () -> {
                  try { Thread.sleep(11); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Weather("Server A", "Sunny");
              };
        Supplier<Weather> fetchWeatherB =
              () -> {
                  try { Thread.sleep(10); } // B est plus rapide
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Weather("Server B", "Mostly Sunny");
              };

        CompletableFuture<Weather> taskA = CompletableFuture.supplyAsync(fetchWeatherA);
        CompletableFuture<Weather> taskB = CompletableFuture.supplyAsync(fetchWeatherB);

        CompletableFuture.anyOf(taskA, taskB)
              .thenAccept(System.out::println)
              .join();

        // On peut voir l'état des deux CompletableFutures après anyOf
        System.out.println("taskA = " + taskA);
        System.out.println("taskB = " + taskB);
    }
}

Observation: Even if B is faster (10ms vs 11ms), there is no guarantee that anyOf always returns B first. The JVM and the scheduler can make different choices. The final state of taskA and taskB can be Done or even pending depending on the timing.

4.8 Demo: compose quotation and weather recovery

package org.paumard.several;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class C_ReadingSeveralTasks {

    record TravelPage(Quotation quotation, Weather weather) { }
    record Weather(String server, String weather) { }
    record Quotation(String server, int amount) { }

    public static void run() throws InterruptedException {

        Random random = new Random();

        List<Supplier<Weather>> weatherTasks = buildWeatherTasks(random);
        List<Supplier<Quotation>> quotationTasks = buildQuotationTasks(random);

        // === Météo : anyOf ===
        List<CompletableFuture<Weather>> weatherCFs = new ArrayList<>();
        for (Supplier<Weather> task : weatherTasks) {
            weatherCFs.add(CompletableFuture.supplyAsync(task));
        }

        // === Devis : allOf ===
        List<CompletableFuture<Quotation>> quotationCFs = new ArrayList<>();
        for (Supplier<Quotation> task : quotationTasks) {
            quotationCFs.add(CompletableFuture.supplyAsync(task));
        }

        CompletableFuture<Void> allOfQuotations =
              CompletableFuture.allOf(quotationCFs.toArray(CompletableFuture[]::new));

        CompletableFuture<Quotation> bestQuotationCF =
              allOfQuotations.thenApply(
                    v -> quotationCFs.stream()
                          .map(CompletableFuture::join)
                          .min(Comparator.comparing(Quotation::amount))
                          .orElseThrow()
              );

        // === Approche 1 : thenCombine ===
        CompletableFuture<TravelPage> pageCompletableFuture =
              bestQuotationCF.thenCombine(
                    CompletableFuture.anyOf(weatherCFs.toArray(CompletableFuture[]::new))
                          .thenApply(o -> (Weather) o),
                    TravelPage::new
              );
        pageCompletableFuture.thenAccept(System.out::println).join();

        // === Approche 2 : thenCompose ===
        CompletableFuture<TravelPage> pageCompletableFuture2 =
              bestQuotationCF.thenCompose(
                    quotation ->
                          CompletableFuture.anyOf(weatherCFs.toArray(CompletableFuture[]::new))
                                .thenApply(o -> (Weather) o)
                                .thenApply(weather -> new TravelPage(quotation, weather))
              );
        pageCompletableFuture2.thenAccept(System.out::println).join();
    }
}

More concise version with Stream API:

package org.paumard.several;

import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class D_ComposingTasks {

    record TravelPage(Quotation quotation, Weather weather) { }
    record Weather(String server, String weather) { }
    record Quotation(String server, int amount) { }

    public static void run() throws InterruptedException {

        Random random = new Random();

        // Conversion directe en tableau avec Stream
        CompletableFuture<Weather>[] weatherCFs = buildWeatherTasks(random).stream()
              .map(CompletableFuture::supplyAsync)
              .toArray(CompletableFuture[]::new);

        CompletableFuture<Weather> weatherCF =
              CompletableFuture.anyOf(weatherCFs)
                    .thenApply(weather -> (Weather) weather);

        CompletableFuture<Quotation>[] quotationCFS = buildQuotationTasks(random).stream()
              .map(CompletableFuture::supplyAsync)
              .toArray(CompletableFuture[]::new);

        CompletableFuture<Void> allOf = CompletableFuture.allOf(quotationCFS);

        CompletableFuture<Quotation> bestQuotationCF = allOf.thenApply(
              v -> Arrays.stream(quotationCFS)
                    .map(CompletableFuture::join)
                    .min(Comparator.comparing(Quotation::amount))
                    .orElseThrow()
        );

        // Composition finale : thenCompose
        CompletableFuture<Void> done =
        bestQuotationCF.thenCompose(
              quotation ->
                    weatherCF.thenApply(weather -> new TravelPage(quotation, weather)))
              .thenAccept(System.out::println);
        done.join();
    }
}

4.9 Module 4 Summary

This module taught you two essential patterns for your asynchronous pipelines:

1. Launch several asynchronous tasks and group the results in different ways:

• anyOf: retrieve only the first result.
• allOf: Have a signal that all tasks are completed.

2. Compose two asynchronous tasks in a non-blocking way with thenCompose.

You now have a good overview of the patterns offered by this API. There are two remaining topics: choice of threads and exception handling.

5. Control which thread executes a task

5.1 Introduction and Agenda

All asynchronous tasks launched so far were running in another thread without you having control. However, there are many cases where you need to control which thread executes which task.

Example: your application accesses an online service and must then query your database. For resource balancing reasons, you want to allocate 10 threads for web access and only 4 for database access. This module teaches you exactly how to do that.

Module Agenda:

1. How the CompletableFuture API chooses default threads.
2. How to take control to impose which thread or thread pool executes which task.

5.2 Default threads used by the CompletionStage API

The default thread pool is common ForkJoinPool.

Common ForkJoinPool characteristics:

• Launched automatically when your application starts.
• Used for several things, including your parallel stream calculations.
• The number of threads is based on the number of cores in your CPU.

Difference between ExecutorService and ForkJoinPool:

This work stealing is a common pattern in concurrent programming, implemented by the ForkJoinPool in the JDK.

Questions to ask:

• For asynchronous I/O operations, you should give more threads than cores, because I/O blocks threads.
• You may need to balance your application by giving different number of threads to disk, network, database operations.
• You may even need to give certain tasks to a very specific thread (eg: updating a GUI — GUIs generally only allow updates from specific threads).

5.3 Choosing an Executor for your asynchronous tasks

Naming convention in the CompletableFuture API:

Thread control example:

// Sans async : email, writeToDB, updateGUI s'exécutent dans le même thread
// que celui qui a produit le résultat de supplyAsync
CompletableFuture<Void> pipeline =
    CompletableFuture.supplyAsync(fetchQuotation)      // ForkJoinPool
        .thenApply(q -> sendEmail(q))                  // même thread
        .thenApply(r -> writeToDatabase(r))             // même thread
        .thenRun(() -> updateGUI());                    // même thread

// Avec async : chaque étape peut changer de thread
CompletableFuture<Void> pipeline2 =
    CompletableFuture.supplyAsync(fetchQuotation, webExecutor)
        .thenApplyAsync(q -> sendEmail(q), emailExecutor)
        .thenApplyAsync(r -> writeToDatabase(r), dbExecutor)
        .thenRunAsync(() -> updateGUI(), guiExecutor);

Note: Executor is the parent interface of ExecutorService. It is simpler (functional interface with a single execute(Runnable) method). Since ExecutorService extends Executor, you can always pass an ExecutorService where an Executor is expected.

Summary table of methods with Executor:

5.4 Demo: specifying Executors for tasks

package org.paumard.thread;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.function.Supplier;

public class D_ComposingTasks {

    record TravelPage(Quotation quotation, Weather weather) { }
    record Weather(String server, String weather) { }
    record Quotation(String server, int amount) { }

    // ThreadFactories personnalisées pour nommer les threads
    private static int quotationThreadIndex = 0;
    private static ThreadFactory quotationThreadFactory =
          r -> new Thread(r, "Quotation-" + quotationThreadIndex++);

    private static int weatherThreadIndex = 0;
    private static ThreadFactory weatherThreadFactory =
          r -> new Thread(r, "Weather-" + weatherThreadIndex++);

    private static int minThreadIndex = 0;
    private static ThreadFactory minThreadFactory =
          r -> new Thread(r, "Min-" + minThreadIndex++);

    public static void run() throws InterruptedException {

        // Trois executors distincts
        ExecutorService quotationExecutor =
              Executors.newFixedThreadPool(4, quotationThreadFactory);
        ExecutorService weatherExecutor =
              Executors.newFixedThreadPool(4, weatherThreadFactory);
        ExecutorService minExecutor =
              Executors.newFixedThreadPool(1, minThreadFactory);

        Random random = new Random();

        List<Supplier<Weather>> weatherTasks = buildWeatherTasks(random);
        List<Supplier<Quotation>> quotationTasks = buildQuotationTasks(random);

        // Météo dans weatherExecutor
        List<CompletableFuture<Weather>> weatherCFs = new ArrayList<>();
        for (Supplier<Weather> weatherTask : weatherTasks) {
            CompletableFuture<Weather> weatherCF =
                  CompletableFuture.supplyAsync(weatherTask, weatherExecutor);
            weatherCFs.add(weatherCF);
        }

        CompletableFuture<Weather> anyOfWeather =
              CompletableFuture
                    .anyOf(weatherCFs.toArray(CompletableFuture[]::new))
                    .thenApply(weather -> (Weather) weather);

        // Devis dans quotationExecutor
        List<CompletableFuture<Quotation>> quotationCFs = new ArrayList<>();
        for (Supplier<Quotation> quotationTask : quotationTasks) {
            CompletableFuture<Quotation> quotationCF =
                  CompletableFuture.supplyAsync(quotationTask, quotationExecutor);
            quotationCFs.add(quotationCF);
        }

        CompletableFuture<Void> allOfQuotations =
              CompletableFuture.allOf(quotationCFs.toArray(CompletableFuture[]::new));

        // Calcul du min dans minExecutor (thenApplyAsync avec executor)
        CompletableFuture<Quotation> bestQuotationCF =
              allOfQuotations.thenApplyAsync(
                    v -> {
                        System.out.println("AllOf then apply " + Thread.currentThread());
                        return quotationCFs.stream()
                              .map(CompletableFuture::join)
                              .min(Comparator.comparing(Quotation::amount))
                              .orElseThrow();
                    },
                    minExecutor
              );

        // Composition finale
        CompletableFuture<Void> done =
              bestQuotationCF.thenCompose(
                          quotation ->
                                anyOfWeather.thenApply(
                                      weather -> new TravelPage(quotation, weather)))
                    .thenAccept(System.out::println);
        done.join();

        // Libérer les ressources
        quotationExecutor.shutdown();
        weatherExecutor.shutdown();
        minExecutor.shutdown();
    }
    // ...
}

Demo observations:

• Without specifying an executor, everything runs in the common ForkJoinPool (threads named ForkJoinPool.commonPool-worker-X).
• With executors specified, threads are named according to the ThreadFactory: Quotation-0, Weather-1, Min-0, etc.
• thenApplyAsync(fn, executor): the function is executed in the thread of the specified executor.

5.5 Module 5 Summary

This module taught you:

1. Additional patterns and Additional methods to specify which executor (which thread) will execute the different stages of your asynchronous pipeline.
2. The naming convention is very strong and helps you understand what the method does.
3. The last topic: exception handling.

6. Report and recover from errors

6.1 Introduction and agenda

The CompletableFuture API is designed for input/output operations: disk access, network, databases. These operations are destined to fail at one point or another. A disk may be inaccessible, a network may be unavailable, a database may be offline.

In synchronous Java, you use try-catch blocks. But in an asynchronous context, certain exceptions can be caught outside your code, preventing you from handling them with a classic try-catch.

This module presents the solutions offered by the CompletableFuture API for reporting and recovering exceptions.

6.2 What happens when a task fails with an exception?

Fundamental rule:

If a task completes exceptionally (it threw an exception instead of producing a result), then all downstream tasks also complete exceptionally.

Pipeline example:

getQuotations()  →  saveToDatabase()  →  mapToValues()
                                                ↓
getWeatherForecast() → → → → → → → → → updateGUI()

• If getQuotations() fails → saveToDatabase() and mapToValues() also fail.
• If getWeatherForecast() fails → updateGUI() also fails.
• But tasks that don’t depend on the failed task can complete normally.

If you do not write specific code to handle these exceptions, your data processing pipeline may fail in an uncontrolled manner. The CompletableFuture API gives you several ways to handle this.

6.3 Log and recover exceptionally

exceptionally(Function<Throwable, T>): first exception handling pattern.

CompletableFuture<Weather> weatherCF =
    CompletableFuture.supplyAsync(fetchWeather)
        .exceptionally(exception -> {
            System.out.println("Exception weather: " + exception);
            return new Weather("Unknown", "Unknown"); // valeur par défaut
        });

Behavior:

This is the pattern to use to recover from an error by providing a default value.

6.4 Handle exceptions with whenComplete and handle

whenComplete(BiConsumer<T, Throwable>) :

CompletableFuture<Quotation> quotationCF =
    CompletableFuture.supplyAsync(fetchQuotation)
        .whenComplete((quotation, exception) -> {
            if (exception != null) {
                System.out.println("Exception: " + exception);
                // ⚠ whenComplete NE récupère PAS l'exception
                // Les tâches aval échoueront aussi
            }
        });

• Takes a BiConsumer<T, Throwable> — does not produce a value.
• One of the two parameters will be null: if exception → quotation == null; otherwise → exception == null.
• If the previous task fails, whenComplete also exceptionally completes with the same exception.
• Not designed to recover from an exception. Useful for logging.

handle(BiFunction<T, Throwable, U>) :

CompletableFuture<Quotation> quotationCF =
    CompletableFuture.supplyAsync(fetchQuotation)
        .handle((quotation, exception) -> {
            if (exception == null) {
                return quotation; // pas d'exception, retourner la valeur normale
            } else {
                System.out.println("exception = " + exception);
                return new Quotation("Unknown", 1_000); // valeur de remplacement
            }
        });

• Takes a BiFunction<T, Throwable, U> — produces a value.
• One of the two parameters will be null (same rule as whenComplete).
• Can recover from an exception by providing a value.
• Combines exception handling and value mapping.

Comparison table of the three methods:

6.5 Demo: exception handling in the TravelPage application

package org.paumard.exception;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class E_ComposingTasks {

    record TravelPage(Quotation quotation, Weather weather) { }
    record Weather(String server, String weather) { }
    record Quotation(String server, int amount) { }

    public static void run() throws InterruptedException {

        Random random = new Random();

        List<Supplier<Weather>> weatherTasks = buildWeatherTasks(random);
        List<Supplier<Quotation>> quotationTasks = buildQuotationTasks(random);

        // === Météo avec récupération via exceptionally ===
        List<CompletableFuture<Weather>> weatherCFs = new ArrayList<>();
        for (Supplier<Weather> weatherTask : weatherTasks) {
            CompletableFuture<Weather> weatherCF =
                  CompletableFuture.supplyAsync(weatherTask)
                              .exceptionally(e -> {
                                  System.out.println("e = " + e);
                                  return new Weather("Unknown", "Unknown");
                              });
            weatherCFs.add(weatherCF);
        }

        CompletableFuture<Weather> anyOfWeather =
              CompletableFuture
                    .anyOf(weatherCFs.toArray(CompletableFuture[]::new))
                    .thenApply(weather -> (Weather) weather);

        // === Devis avec récupération via handle ===
        List<CompletableFuture<Quotation>> quotationCFs = new ArrayList<>();
        for (Supplier<Quotation> quotationTask : quotationTasks) {
            CompletableFuture<Quotation> quotationCF =
                  CompletableFuture.supplyAsync(quotationTask)
                        .handle(
                              (quotation, exception) -> {
                                  if (exception == null) {
                                      return quotation;
                                  } else {
                                      System.out.println("exception = " + exception);
                                      return new Quotation("Unknown", 1_000); // valeur pénalisée
                                  }
                              }
                        );
            quotationCFs.add(quotationCF);
        }

        CompletableFuture<Void> allOfQuotations =
              CompletableFuture.allOf(quotationCFs.toArray(CompletableFuture[]::new));

        CompletableFuture<Quotation> bestQuotationCF =
              allOfQuotations.thenApply(
                    v -> quotationCFs.stream()
                          .map(CompletableFuture::join)
                          .min(Comparator.comparing(Quotation::amount))
                          .orElseThrow()
              );

        CompletableFuture<Void> done =
              bestQuotationCF.thenCompose(
                          quotation ->
                                anyOfWeather.thenApply(
                                      weather -> new TravelPage(quotation, weather)))
                    .thenAccept(System.out::println);
        done.join();
    }

    // Serveur B météo simule une panne
    private static List<Supplier<Weather>> buildWeatherTasks(Random random) {
        Supplier<Weather> fetchWeatherA =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Weather("Server A", "Sunny");
              };
        Supplier<Weather> fetchWeatherB =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  // Simule une panne
                  throw new RuntimeException(
                        new IOException("Weather server B unavailable"));
              };
        Supplier<Weather> fetchWeatherC =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Weather("Server C", "Almost Sunny");
              };
        return List.of(fetchWeatherA, fetchWeatherB, fetchWeatherC);
    }

    // Serveur A devis simule une panne
    private static List<Supplier<Quotation>> buildQuotationTasks(Random random) {
        Supplier<Quotation> fetchQuotationA =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  // Simule une panne
                  throw new RuntimeException(
                        new IOException("Quotation server A unavailable"));
              };
        Supplier<Quotation> fetchQuotationB =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Quotation("Server B", random.nextInt(30, 70));
              };
        Supplier<Quotation> fetchQuotationC =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  return new Quotation("Server C", random.nextInt(40, 80));
              };
        return List.of(fetchQuotationA, fetchQuotationB, fetchQuotationC);
    }
}

Demo strategy:

• fetchWeatherB always throws a RuntimeException (weather server B unavailable).
• fetchQuotationA always throws a RuntimeException (quotation server A unavailable).
• For weather: exceptionally catches the exception and returns new Weather("Unknown", "Unknown") as default.
• For quotes: handle catches the exception and returns new Quotation("Unknown", 1_000) — a quote “penalized” with a high amount (1000) so that it is never selected as the best quote.
• The overall pipeline continues to work despite partial failures.

6.6 Module 6 Summary

This module taught you the three patterns of exception handling:

1. exceptionally(Function<Throwable, T>): only called if an exception is thrown. Can recover by providing a default value.

2. whenComplete(BiConsumer<T, Throwable>): takes a BiConsumer (no return value). One of the two parameters will be null. Does not catch the exception — downstream tasks still fail.

3. handle(BiFunction<T, Throwable, U>): takes a BiFunction (produces a value). One of the two parameters will be null. Can retrieve by providing a value. Combines exception management and value transformation.

Each of these three methods exists in async versions (9 additional methods in the CompletableFuture API).

7. Closing Remarks

7.1 Introduction and agenda

This last module summarizes everything that has been covered about the CompletableFuture API and its many methods. The objective is to give you the keys to find your way in this complex API.

Topics covered:

• Supported task templates.
• How to chain tasks.
• Managing exceptions.
• The choice of threads.
• Performance.

7.2 Patterns to create an asynchronous task

Basic pattern:

// Lancer une tâche asynchrone produisant un résultat
CompletableFuture<T> cf = CompletableFuture.supplyAsync(supplier);

// Avec contrôle du thread
CompletableFuture<T> cf = CompletableFuture.supplyAsync(supplier, executor);

Factory methods for multiple concurrent tasks:

// allOf : complété quand TOUTES les tâches sont terminées
// Retourne CompletableFuture<Void> — le Void est toujours null !
CompletableFuture<Void> allOf = CompletableFuture.allOf(cf1, cf2, cf3);

// anyOf : complété quand AU MOINS UNE tâche est terminée
// Retourne CompletableFuture<Object> — cast nécessaire
CompletableFuture<Object> anyOf = CompletableFuture.anyOf(cf1, cf2, cf3);

7.3 Patterns for modeling tasks

The CompletableFuture API supports three models for your tasks:

Methods of CompletableFuture are named after the abstract method:

thenRun(Runnable)     ← "run" ← Runnable.run()
thenAccept(Consumer)  ← "accept" ← Consumer.accept()
thenApply(Function)   ← "apply" ← Function.apply()

7.4 Patterns for chaining and composing tasks

Three methods to chain a task in memory:

CompletableFuture<Void>  cf.thenRun(runnable)        // pas d'accès au résultat
CompletableFuture<Void>  cf.thenAccept(consumer)     // accès au résultat, pas de retour
CompletableFuture<U>     cf.thenApply(function)      // accès au résultat, retourne valeur

A method for dealing with a long running task (I/O):

// thenCompose : fn doit retourner un CompletableFuture
CompletableFuture<U> cf.thenCompose(fn: T -> CompletableFuture<U>)

Each method exists in 3 versions:

// Version 1 : même thread
cf.thenApply(fn)

// Version 2 : autre thread (common ForkJoinPool)
cf.thenApplyAsync(fn)

// Version 3 : thread de l'executor spécifié
cf.thenApplyAsync(fn, executor)

Which gives 4 × 3 = 12 methods just for chaining one task onto one.

Trigger on the result of TWO tasks:

// thenCombine : combine les résultats de this et other avec biFunction
CompletableFuture<U> cf1.thenCombine(cf2, biFunction)

// thenAcceptBoth : consomme les deux résultats sans retour
CompletableFuture<Void> cf1.thenAcceptBoth(cf2, biConsumer)

// runAfterBoth : exécute runnable quand les deux sont terminées
CompletableFuture<Void> cf1.runAfterBoth(cf2, runnable)

These methods also exist in async versions with or without executor.

7.5 Patterns to handle exceptions

Rule: if a task throws an exception, all tasks that depend on it also throw an exception.

// exceptionally : récupère l'exception, fournit une valeur par défaut
CompletableFuture<T> cf.exceptionally(fn: Throwable -> T)

// handle : traite exception ET valeur, peut récupérer
CompletableFuture<U> cf.handle(biFunction: (T, Throwable) -> U)
// ⚠ L'un des deux paramètres est null

// whenComplete : observe exception ET valeur, NE récupère PAS
CompletableFuture<T> cf.whenComplete(biConsumer: (T, Throwable) -> void)
// ⚠ L'un des deux paramètres est null
// ⚠ Si exception, se complète aussi exceptionnellement

These 3 methods each have 3 versions (without async, with async, with async + executor) = 9 additional methods.

7.6 Patterns for better performance

Performance Tips:

1. Do not use the CompletableFuture API for purely in-memory calculations. There is an overhead due to the pipeline. You would decrease performance instead of improving it. This API is designed for I/O pipelines.

2. Precisely identify long I/O tasks vs. in-memory calculations when designing your pipelines.

3. A single executor or several?

• A single executor: simpler and less prone to errors.
• Using the default common ForkJoinPool may not be the best idea — your own executor may be preferable.
• A single executor can also be faster because transferring data from one thread to another has a cost (overhead).

4. If you use multiple executors: carefully choose the number of threads in each executor.

• Too few threads → tasks are waiting, throughput drops.
• Too many threads → context-switching overhead.
• For blocking I/O → more threads than CPU cores is justified.

7.7 Final Course Summary

This course covered the CompletionStage and CompletableFuture API:

• CompletionStage is the interface.
• CompletableFuture is the class that implements this interface.

What you learned:

1. How tasks are modeled: Runnable, Consumer, Function.
2. How to control which thread executes which task and efficiently design your application with executors.
3. How to catch exceptions with exceptionally, whenComplete, handle.

To go further:

• Follow the Project Loom project (OpenJDK) which brings new patterns for long I/O operations: virtual threads and structured concurrency — entirely new notions in the JDK.

Instructor Resources (Jose Paumard):

• Twitter: for the latest Java news
• YouTube: lectures in English and French
• GitHub: source code for examples
• SlideShare: conference slides

8. Demo source code

8.1 Module 2 — A: Accessing Data Asynchronously

File structure:

02/demos/A_Accessing-Data-Asynchronously/src/org/paumard/async/
├── A_RunSynchronousTasks.java   ← Approche synchrone (référence)
├── B_RunExecutorTasks.java      ← Approche ExecutorService
├── C_RunAsyncTasks.java         ← Approche CompletableFuture
└── RunAll.java                  ← Lance les 3 approches en séquence

RunAll.java — Comparison of the 3 approaches:

package org.paumard.async;

import java.util.concurrent.ExecutionException;

public class RunAll {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        A_RunSynchronousTasks.run();   // ~300ms
        B_RunExecutorTasks.run();      // ~120ms
        C_RunAsyncTasks.run();         // ~120ms
    }
}

Typical output:

Best quotation [SYNC ] = Quotation[server=Server A, amount=45] (312ms)
Best quotation [ES   ] = Quotation[server=Server B, amount=38] (118ms)
Best quotation [ASYNC] = Quotation[server=Server B, amount=41] (115ms)

8.2 Module 3 — B: Triggering a Task

File structure:

03/demos/B_Triggering-a-Task/src/org/paumard/chaining/
└── A_TriggerTasks.java   ← thenAccept, ConcurrentLinkedDeque

8.3 Module 4 — C: Splitting a Result

File structure:

04/demos/C_Splitting-a-Result/src/org/paumard/several/
├── A_ReadingSeveralTasks.java        ← allOf + thenApply pour extraire le min
├── B_ReadingOneOfSeveralTasks.java   ← anyOf
├── C_ReadingSeveralTasks.java        ← thenCombine + thenCompose
├── D_ComposingTasks.java             ← Version Stream API, plus concise
└── E_AnyOf.java                      ← Comportement détaillé d'anyOf

C_RunAsyncTasks.java (module 5) — Observing threads:

package org.paumard.thread;

// ... imports ...

public class C_RunAsyncTasks {

    record Quotation(String server, int amount) { }

    public static void run() {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        Random random = new Random();

        List<Supplier<Quotation>> quotationTasks = buildQuotationTasks(random);

        // Sans executor spécifié → common ForkJoinPool
        List<CompletableFuture<Quotation>> futures = new ArrayList<>();
        for (Supplier<Quotation> quotationTask : quotationTasks) {
            var future = CompletableFuture.supplyAsync(quotationTask);
            futures.add(future);
        }

        List<Quotation> quotations = new ArrayList<>();
        for (CompletableFuture<Quotation> future : futures) {
            Quotation quotation = future.join();
            quotations.add(quotation);
        }

        Quotation bestQuotation =
              quotations.stream()
                    .min(Comparator.comparing(Quotation::amount))
                    .orElseThrow();

        System.out.println("Best quotation = " + bestQuotation);
    }

    private static List<Supplier<Quotation>> buildQuotationTasks(Random random) {
        Supplier<Quotation> fetchQuotationA =
              () -> {
                  try { Thread.sleep(random.nextInt(80, 120)); }
                  catch (InterruptedException e) { throw new RuntimeException(e); }
                  // Affiche le thread qui exécute cette tâche
                  System.out.println("Quotation A in thread " + Thread.currentThread());
                  return new Quotation("Server A", random.nextInt(40, 60));
              };
        // ... fetchQuotationB, fetchQuotationC similaires
        return List.of(fetchQuotationA, /*...*/null, null);
    }
}

Typical output (without executor specified):

Quotation A in thread Thread[#21,ForkJoinPool.commonPool-worker-1,5,main]
Quotation B in thread Thread[#22,ForkJoinPool.commonPool-worker-2,5,main]
Quotation C in thread Thread[#23,ForkJoinPool.commonPool-worker-3,5,main]
Best quotation = Quotation[server=Server B, amount=35]

8.4 Module 5 — D: Controlling Threads

File structure:

05/demos/D_Controlling-Threads/src/org/paumard/thread/
├── C_RunAsyncTasks.java    ← Observer les threads du ForkJoinPool
└── D_ComposingTasks.java   ← Executors distincts par type de tâche

Typical output with named executors:

WA running in Thread[#21,Weather-0,5,main]
WB running in Thread[#22,Weather-1,5,main]
QA running in Thread[#23,Quotation-0,5,main]
QB running in Thread[#24,Quotation-1,5,main]
QC running in Thread[#25,Quotation-2,5,main]
WC running in Thread[#26,Weather-2,5,main]
AllOf then apply Thread[#27,Min-0,5,main]
TravelPage[quotation=Quotation[server=Server B, amount=38], weather=Weather[server=Server A, weather=Sunny]]

8.5 Module 6 — E: Reporting Errors

File structure:

06/demos/E_Reporting-Errors/src/org/paumard/exception/
└── E_ComposingTasks.java   ← exceptionally + handle pour récupérer des pannes

Typical output with simulated faults:

e = java.util.concurrent.CompletionException: java.io.IOException: Weather server B unavailable
exception = java.util.concurrent.CompletionException: java.io.IOException: Quotation server A unavailable
TravelPage[quotation=Quotation[server=Server C, amount=52], weather=Weather[server=Server A, weather=Sunny]]

9. Appendix: Summary of CompletableFuture Methods

Factory methods (creation)

Chaining methods

Methods for two tasks

Exception handling methods

Naming convention — Async suffix

Each method of chaining and handling exceptions exists in 3 variants:

méthode(fn)            → même thread que la tâche précédente
méthodeAsync(fn)       → thread du common ForkJoinPool
méthodeAsync(fn, exec) → thread de l'executor spécifié

Search Terms

asynchronous · programming · java · backend · architecture · full-stack · web · tasks · agenda · task · methods · patterns · allof · avoid · blocking · completablefuture · completionstage · exception · launch · threads · anyof · api · application · asynchronously