9. Asynchronous Programming

The previous chapter introduced threads: we can execute multiple pieces of code “at the same time,” synchronize them, and assign tasks to a thread pool via an executor ([ExecutorService]). This chapter addresses a related but distinct issue: how to write a program that waits a lot (for the disk, the network, or a database) without tying up threads during these waits.

Java does not have language keywords for this (other languages, such as C# or JavaScript, have the keywords [async] and [await]). It offers two solutions, which complement each other and are both presented in this chapter:

• the [CompletableFuture<T>] class (Java 8, since enhanced) represents an ongoing operation whose subsequent steps are composed using callback methods: “when the result is ready, do this, then that” ([thenApply], [thenCompose], [thenCombine], [allOf]…). No thread waits: it is the completion of the operation that triggers the next step;
• virtual threads ( Java 21), on the other hand, allow you to write blocking, sequential code that reads from top to bottom: when a virtual thread waits for input/output, the JVM detaches it from the system thread that was executing it (its “carrier thread”) and frees that thread for another virtual thread. This makes it possible to have hundreds of thousands of virtual threads for just a few system threads. This is Java’s modern solution to the problem that other languages solve with [async] / [await].

These tools are everywhere today: the following chapters will explore them to access databases, write TCP servers that serve each client via a virtual thread, web clients using [java.net.http.HttpClient], and web services using Spring Boot.

All the examples in this chapter are compact source files ([Async01.java], [Async02.java]…) that are run using [java AsyncNN.java] in the [09-async] folder. Many of them measure execution times using [System.nanoTime()]: the milliseconds displayed vary from one run to another and from one machine to another (the examples were run on a dual-processor machine). It’s the orders of magnitude that matter: 1 second or 10 seconds, 300 ms or 0 ms.

9.1. Why Asynchrony

9.1.1. Operations that wait

A program performs two types of work:

• computational tasks, which occupy the processor: sorting a list, calculating a tax, analyzing a text;
• input/output (I/O), which essentially consists of waiting for a peripheral device: reading a file from disk, sending a request to a web server and waiting for its response, executing a SQL query on a database server.

The orders of magnitude are very different. A processor executes a few billion instructions per second. A read operation on a disk (SSD) takes on the order of one-tenth of a millisecond; a round trip over a local network takes a fraction of a millisecond; a round trip to a server located on the other side of the country takes a few tens of milliseconds; and a somewhat complex SQL request can sometimes take several seconds. During these waits, the processor could have executed millions of instructions.

When a program reads a line over a network connection using a blocking method, the thread performing this read is blocked: it does nothing, but it still exists. A “classic” Java thread—now referred to as a “platform thread”—is an operating system thread: it occupies memory (its stack, on the order of a megabyte), it is recognized by the system, and, most importantly, it is no longer available to do anything else. This isn’t a problem in a small console program that does one thing at a time. It becomes much more of an issue in two situations:

• in a server: each client served by a blocked thread ties up that thread. If the server’s thread pool has N threads and all N threads are performing a blocking operation, there are no threads left available for a new client. A modern server handles thousands of simultaneous requests: this is possible because its waiting operations do not tie up system threads;
• in a graphical user interface: a single thread—the UI thread (Swing’s “Event Dispatch Thread,” the “JavaFX Application Thread”)—handles clicks and redraws windows. If it is blocked by a download, the window freezes.

Asynchrony involves initiating the I/O operation, freeing up the system thread while it is in progress (the hardware and operating system do not need it to transfer data), and then resuming work once the operation is complete, possibly on another system thread:

• In [1], the platform thread T1 initiates the I/O and remains blocked until it finishes;
• in [2], the virtual thread V1 initiates the I/O; JVM detaches it from its parent thread P1, which can then immediately execute other virtual threads. When the I/O completes, JVM reattaches V1 to a free carrier thread, P2 (which could also be P1), and V1 continues its execution immediately after the blocking instruction, as if nothing had happened.

The program [Async01.java] measures the difference over 100 operations, each of which waits for one second (the wait simulates I/O). It uses instructions that we will explain later in this chapter ([Executors.newVirtualThreadPerTaskExecutor], [CompletableFuture]): here, we are primarily interested in the results.

// why asynchrony: 100 operations that each wait one second
// (a wait simulates an I/O: disk read, network request, SQL query…)
import java.lang.management.ManagementFactory;

final int nbOperations = 100;
long start;

void main() throws InterruptedException {
    IO.println(nbOperations + " operations of one second, "
            + Runtime.getRuntime().availableProcessors() + " processors");

    // 1 - one platform thread per operation, each thread is blocked during the wait
    start = System.nanoTime();
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < nbOperations; i++) {
        threads.add(Thread.ofPlatform().start(() -> waitFor(1000)));
    }
    display("1 - one platform thread per operation", jvmThreads());
    for (Thread thread : threads) {
        thread.join();
    }
    display("    end", -1);

    // 2 - the operations are given to a pool of 10 threads, but they block it
    start = System.nanoTime();
    try (ExecutorService pool = Executors.newFixedThreadPool(10)) {
        for (int i = 0; i < nbOperations; i++) {
            pool.submit(() -> waitFor(1000));
        }
        display("2 - pool of 10 threads, blocking waits", jvmThreads());
    } // close() waits for the end of all the operations
    display("    end", -1);

    // 3 - one virtual thread per operation: it releases its carrier thread during the wait
    start = System.nanoTime();
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < nbOperations; i++) {
            executor.submit(() -> waitFor(1000));
        }
        display("3 - one virtual thread per operation", jvmThreads());
    }
    display("    end", -1);

    // 4 - asynchronous waits: no thread waits, a timer completes each operation
    start = System.nanoTime();
    var operations = new CompletableFuture<?>[nbOperations];
    for (int i = 0; i < nbOperations; i++) {
        operations[i] = new CompletableFuture<Void>().completeOnTimeout(null, 1000, TimeUnit.MILLISECONDS);
    }
    display("4 - asynchronous waits (CompletableFuture)", jvmThreads());
    CompletableFuture.allOf(operations).join();
    display("    end", -1);
}

// a blocking wait of [ms] milliseconds
void waitFor(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

// number of platform threads of the JVM, half a second after the operations are started
int jvmThreads() {
    long elapsed = (System.nanoTime() - start) / 1_000_000;
    if (elapsed < 500) {
        waitFor(500 - elapsed);
    }
    return ManagementFactory.getThreadMXBean().getThreadCount();
}

// display of the elapsed time and, if provided (>= 0), of the number of threads
void display(String message, int nbThreads) {
    String threadsInfo = nbThreads < 0 ? "" : String.format(", %3d threads in the JVM", nbThreads);
    long elapsed = (System.nanoTime() - start) / 1_000_000;
    IO.println(String.format("%-44s : %5d ms%s", message, elapsed, threadsInfo));
}

Results:

> java Async01.java
100 operations of one second, 2 processors
1 - one platform thread per operation        :   534 ms, 106 threads in the JVM
    end                                      :  1061 ms
2 - pool of 10 threads, blocking waits       :   501 ms,  16 threads in the JVM
    end                                      : 10021 ms
3 - one virtual thread per operation         :   501 ms,  10 threads in the JVM
    end                                      :  1091 ms
4 - asynchronous waits (CompletableFuture)   :   500 ms,  11 threads in the JVM
    end                                      :  1003 ms

Comments:

• Lines 12–22: the simplest solution—one platform thread per operation. [Thread.ofPlatform().start(task)] (Java 21) creates and starts a thread, just like [new Thread(task).start()]. Each thread executes [Thread.sleep(1000)]: it is blocked for one second. The 100 waits occur simultaneously, and the entire process takes exactly one second (line 4 of the results), but at the cost of 100 additional system threads: JVM has 106 of them (line 3). With 10,000 operations, 10,000 system threads would be required;
• Lines 24–32: The operations are assigned to a pool of 10 threads ([Executors.newFixedThreadPool(10)]), a reasonable size for computations. But here, these threads aren’t computing—they’re idle. The operations wait for a free thread, in batches of 10, and the whole process takes… ten seconds (line 6). This is known as “thread pool starvation,” the most common flaw in servers that mix thread pools with blocking I/O. The method [close()] of an executor (Java 19), called upon exit from [try], rejects new tasks and waits for the ones that have been submitted to finish;
• Lines 34–42: One virtual thread per operation. [Executors.newVirtualThreadPerTaskExecutor()] (Java 21) returns an executor that creates a new virtual thread for each submitted task. The task code is the same as in Step 1: a blocking [Thread.sleep(1000)]. However, a blocked virtual thread does not tie up a system thread: the JVM detaches it from its carrier thread. The 100 waits last a total of one second (line 8), and the JVM counts only 10 system threads (line 7): its service threads, and the two carrier threads—one per processor—that executed the 100 virtual threads. The virtual threads are not included in this count: they are objects of JVM, not system threads;
• lines 44–52: the same wait with no threads waiting. [new CompletableFuture<Void>()] creates a “pending” operation, and [completeOnTimeout(value, timeout, unit)] (Java 9) requests that it be completed with the specified value once the timeout expires. A single timer from JVM manages the 100 timeouts. [CompletableFuture.allOf(...)] returns an operation that completes when all others have finished, and [join()] waits for it to finish. The entire process lasts one second (line 10), with only one additional thread compared to step 3—the timer thread (line 9);
• lines 55–62: [Thread.sleep] can throw the checked exception [InterruptedException] if the thread is interrupted while sleeping (we’ll discuss this further in the section on cancellation); the [waitFor] method intercepts it so that the lambdas on lines 16, 28, and 38 do not have to handle it;
• Lines 64–71: [ManagementFactory.getThreadMXBean().getThreadCount()] returns the number of live platform threads for JVM. The class [ManagementFactory] belongs to the module [java.management]: a compact source file automatically imports only the module [java.base], hence the import on line 3.

Let’s remember the lesson: to wait, you must not block any system threads. Asynchrony doesn’t make I/O faster; it allows you to handle a large number of I/O operations simultaneously using few system threads, while keeping the threads the application needs available. Java offers two approaches for this: creating [CompletableFuture] (Step 4), or writing ordinary blocking code executed by virtual threads (Step 3).

9.1.2. Several Successive Models

Java has offered several ways to write code that waits without blocking. The older ones are still found in existing code, so it’s useful to recognize them.

Pattern	Introduction	Principle
[Future] and Runtimes	Java 5 (2004)	[ExecutorService.submit] assigns a task to a thread pool and returns a [Future<T>], which represents the upcoming result. But the only way to obtain this result is to wait, blocking, with [get()].
Callback methods of NIO.2	Java 7 (2011)	Asynchronous channels ([AsynchronousSocketChannel], [AsynchronousFileChannel]) initiate an operation and receive a callback object ([CompletionHandler]) whose method [completed] or [failed] is called by a thread from a pool once the operation is complete.
[CompletableFuture]	Java 8 (2014)	a [Future] that you can terminate yourself and whose sequence is composed of callback methods ([thenApply], [thenCompose], [exceptionally]…); reactive streams [java.util.concurrent.Flow] (Java 9) are an extension of this for sequences of values. Reactive libraries (Project Reactor, used by Spring WebFlux, RxJava) are based on the same concepts.
virtual threads	Java 21 (2023)	of ordinary blocking code, executed by virtual threads: JVM frees the system thread during waits. This is the recommended model today for new applications.
structured concurrency	Preview of Java 21 through Java 25	[StructuredTaskScope] organizes a group of subtasks executed by virtual threads as a single block of code, with shared error handling and cancellation (see the end of the chapter).

Callback methods have a drawback: the code is fragmented. What naturally reads as “connect, then read, then display” becomes a method that initiates the connection, a callback method that initiates the read operation, and another callback method that displays the result; variables must be shared among these methods, exceptions are caught at different locations, and a simple loop (“read as long as there is data”) becomes a callback method that recursively calls itself. [CompletableFuture] makes this fragmentation more readable by chaining the steps together; virtual threads make it disappear: asynchronous code takes the form of synchronous code, because it is synchronous code.

The program [Async02.java] implements the same client, TCP, using two extreme approaches: first, using the callback methods of NIO.2, and then using blocking code executed by a virtual thread. It also contains a small server, run by a thread, that responds to two clients (network programming will be covered in detail in Chapter 12; for now, it suffices to know that a client connects to the server and reads the message the server sends it).

// the same TCP client written with callback methods (NIO.2, Java 7), then with a virtual thread (Java 21)
final int port = 27500;
final InetAddress local = InetAddress.getLoopbackAddress();
final ByteBuffer buffer = ByteBuffer.allocate(1024);
final CountDownLatch end = new CountDownLatch(1);

void main() throws IOException, InterruptedException {
    // a minimal server, executed by a thread: it serves two clients then stops
    var listener = new ServerSocket(port, 10, local);
    Thread server = Thread.ofPlatform().name("server").start(() -> {
        try (listener) {
            for (int i = 1; i <= 2; i++) {
                try (Socket service = listener.accept()) {
                    // the server takes its time to answer
                    Thread.sleep(500);
                    byte[] message = ("Hello client no. " + i).getBytes(StandardCharsets.UTF_8);
                    service.getOutputStream().write(message);
                }
            }
        } catch (IOException | InterruptedException e) {
            display("server: " + e);
        }
    });
    display("start of the program");

    // 1 - the "callback methods" style: AsynchronousSocketChannel and CompletionHandler
    var client = AsynchronousSocketChannel.open();
    // we start the connection, giving the object to call back when it is done
    client.connect(new InetSocketAddress(local, port), client, new Connected());
    display("callbacks: connection started, the program continues");
    // the main thread has nothing left to do: it waits for the signal of the last callback method
    end.await();

    // 2 - the blocking style, executed by a virtual thread: the code reads from top to bottom
    Thread virtual = Thread.ofVirtual().start(() -> {
        try (var client2 = new Socket(local, port)) {
            display("virtual thread: connected, read started");
            byte[] bytes = new byte[1024];
            int nbBytes = client2.getInputStream().read(bytes);
            String message = new String(bytes, 0, nbBytes, StandardCharsets.UTF_8);
            display("virtual thread: received [" + message + "]");
        } catch (IOException e) {
            display("virtual thread: " + e);
        }
    });
    virtual.join();
    server.join();
}

// callback method no. 1: the connection is made (or has failed)
class Connected implements CompletionHandler<Void, AsynchronousSocketChannel> {
    @Override
    public void completed(Void nothing, AsynchronousSocketChannel channel) {
        display("callbacks: connected, read started");
        // we start reading the server's answer with another callback object
        channel.read(buffer, channel, new Received());
    }

    @Override
    public void failed(Throwable e, AsynchronousSocketChannel channel) {
        display("callbacks: connection failed: " + e);
        end.countDown();
    }
}

// callback method no. 2: data has arrived
class Received implements CompletionHandler<Integer, AsynchronousSocketChannel> {
    @Override
    public void completed(Integer nbBytes, AsynchronousSocketChannel channel) {
        display("callbacks: received [" + new String(buffer.array(), 0, nbBytes, StandardCharsets.UTF_8) + "]");
        try {
            channel.close();
        } catch (IOException e) {
            display("callbacks: " + e);
        }
        // we notify the main thread
        end.countDown();
    }

    @Override
    public void failed(Throwable e, AsynchronousSocketChannel channel) {
        display("callbacks: read failed: " + e);
        end.countDown();
    }
}

// display prefixed by the thread executing it
// (name of a platform thread, description of a virtual thread)
void display(String message) {
    Thread thread = Thread.currentThread();
    IO.println("[" + (thread.isVirtual() ? thread.toString() : thread.getName()) + "] " + message);
}

Results:

> java Async02.java
[main] start of the program
[main] callbacks: connection started, the program continues
[InnocuousThread-4] callbacks: connected, read started
[InnocuousThread-2] callbacks: received [Hello client no. 1]
[VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1] virtual thread: connected, read started
[VirtualThread[#28]/runnable@ForkJoinPool-1-worker-2] virtual thread: received [Hello client no. 2]

Comments:

• Lines 8–23: the server. It listens on port 27500 of the local machine ([InetAddress.getLoopbackAddress()], i.e., 127.0.0.1), accepts two connections one after the other, and sends each client a welcome message after half a second. It is executed by a platform thread named [server]. The form [try (listener)] (Java 9) will close the [ServerSocket] declared on line 9 at the end of the block;
• lines 26–32: the client writes using callback methods:
  • line 27: a *[AsynchronousSocketChannel]* is a TCP channel whose operations are asynchronous;
  • line 29: [connect] initiates the connection and immediately returns control (line 3 of the results). Its last two parameters are an arbitrary object that the callback method will receive (here, the channel itself) and the callback object, of type [Connected];
  • line 32: the main thread has nothing left to do. To ensure the program does not terminate before the exchange is complete, it waits for a signal from the last callback method: a [CountDownLatch] initialized to 1 (line 5), whose method [await] blocks the thread that calls it until another thread calls its method [countDown];
• lines 50–64: The connection’s callback object implements the [CompletionHandler<V, A>] interface, where [V] is the result type of the operation ([Void] for a connection) and [A] is the type of the passed-in object. Its method [completed] is called if the operation succeeds, and its method [failed] is called if it fails. Once connected, reading is initiated with [read] by specifying a second callback object of type [Received] (line 56);
• lines 66–85: the read callback object receives the number of bytes read, displays the message, closes the channel, and releases the main thread. The classes [Connected] and [Received] are internal classes of the implicit class of the compact file: they have access to its fields [buffer] and [end] and to its method [display];
• the results show that the callback methods are executed by threads other than the main thread (threads from the JVM pool dedicated to asynchronous channels, named [InnocuousThread-N]);
• lines 34–46: the same client, written using a standard [Socket] and blocking operations ([new Socket(...)] connects, [read] waits for data), executed by a virtual thread created by [Thread.ofVirtual().start(task)] (Java 21). The code reads from top to bottom like synchronous code; the variables are ordinary local variables; errors are handled by a single [try / catch]; there are no callback methods or signals to manage;
• lines 87–92: Each display is prefixed with the thread that executes it. For a virtual thread, [toString()] returns a description in the form [VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1]: the virtual thread number, its state, and its current host thread. Lines 6 and 7 of the results show the same virtual thread No. 28, carried first by [worker-1] and then by [worker-2]: while waiting for the server’s response, it was detached from its first carrier and then reattached to another.

It is not necessary to retain the [CompletionHandler] interface. If you must use a API that only offers this model, convert it to [CompletableFuture] (we’ll do this in [Async17.java]), or use its blocking version on a virtual thread.

9.2. Upcoming tasks: Future and CompletableFuture

A [Future<T>] (interface of the [java.util.concurrent] package, Java 5) is an object that represents the result—of type [T]—of an ongoing or completed operation. We encountered it in the previous chapter: this is what the [submit] method of an executor returns. You can query it (has it finished? was it successful?), cancel it ([cancel]), and wait for its result ([get]), but nothing else: the only way to use the result is to block a thread until it is available.

The [CompletableFuture<T>] class (Java 8) implements [Future<T>] and adds two essential capabilities:

• you can terminate it yourself, with a result ([complete]) or an exception ([completeExceptionally]);
• you can attach a follow-up operation to it—a function that will be executed when the result is available ([thenApply], [thenAccept], [thenCompose]…), which produces a new [CompletableFuture]. In this way, chains of operations are constructed without any threads having to wait. These methods are declared by the [CompletionStage<T>] interface, which [CompletableFuture] also implements.

A [CompletableFuture] is not a thread. It may represent a process executed by a thread of an executor (as is the case with those created by [supplyAsync]), but many are not executed by any thread: the one in step 4 of [Async01.java] is simply “completed” by a timer after one second, and the one returned by [HttpClient.sendAsync] is completed when the server’s response arrives. It is a promise of a result.

The program [Async03.java] demonstrates the main ways to create, query, and wait for a [CompletableFuture]:

// CompletableFuture: creation, waiting, state, result, continuation
void main() throws InterruptedException {
    // 1 - supplyAsync gives a processing to an executor and returns a CompletableFuture<T>
    CompletableFuture<Long> f1 = CompletableFuture.supplyAsync(() -> {
        // a computation of a few tenths of a second
        long sum = 0;
        for (int i = 1; i <= 200_000_000; i++) {
            sum += i % 7;
        }
        return sum;
    });
    IO.println("f1 started, state: " + f1.state());
    // join() waits for the end of the processing, blocking the current thread, and returns its result
    long result = f1.join();
    IO.println("f1 finished, state: " + f1.state() + ", result: " + result);

    // 2 - a CompletableFuture completed by a timer after 300 ms, without occupying a thread
    CompletableFuture<Void> f2 = new CompletableFuture<Void>()
            .completeOnTimeout(null, 300, TimeUnit.MILLISECONDS);
    IO.println("f2 started, state: " + f2.state());
    f2.join();
    IO.println("f2 finished, state: " + f2.state());

    // 3 - CompletableFutures already completed, successfully or with a failure
    CompletableFuture<Integer> f3 = CompletableFuture.completedFuture(42);
    CompletableFuture<Integer> f4 = CompletableFuture.failedFuture(new IllegalStateException("failure"));
    IO.println("f3: state " + f3.state() + ", result " + f3.resultNow()
            + " ; f4: state " + f4.state() + ", exception " + f4.exceptionNow());

    // 4 - a processing that fails: join() throws a CompletionException that wraps its exception
    CompletableFuture<Integer> f5 = CompletableFuture.supplyAsync(() -> Integer.parseInt("abc"));
    try {
        f5.join();
    } catch (CompletionException e) {
        IO.println("f5: state " + f5.state() + ", join() throws " + e.getClass().getSimpleName()
                + " containing " + e.getCause());
    }

    // 5 - get(), inherited from the Future interface, wraps the exception in an ExecutionException
    try {
        f5.get();
    } catch (ExecutionException e) {
        IO.println("f5: get() throws " + e.getClass().getSimpleName()
                + " containing " + e.getCause().getClass().getSimpleName());
    }

    // 6 - a continuation: what to do when the CompletableFuture is completed
    CompletableFuture<Integer> f6 = f2.thenApply(nothing -> 2 * 21);
    CompletableFuture<Void> f7 = f6.thenAccept(value -> IO.println("f7: the continuation received " + value));
    f7.join();

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

Results:

> java Async03.java
f1 started, state: RUNNING
f1 finished, state: SUCCESS, result: 599999998
f2 started, state: RUNNING
f2 finished, state: SUCCESS
f3: state SUCCESS, result 42 ; f4: state FAILED, exception java.lang.IllegalStateException: failure
f5: state FAILED, join() throws CompletionException containing java.lang.NumberFormatException: For input string: "abc"
f5: get() throws ExecutionException containing NumberFormatException
f7: the continuation received 42
f1: isDone=true, f5: isDone=true, isCompletedExceptionally=true, isCancelled=false

Comments:

• Lines 4–11: [CompletableFuture.supplyAsync(supplier)] passes a function with no parameters ([Supplier<T>]) to an executor and immediately returns a [CompletableFuture<T>] representing its execution. The function returns a [long], so we obtain a [CompletableFuture<Long>]. Without an explicit executor, the common pool for JVM and [ForkJoinPool.commonPool()] executes the function; we’ll see later how to choose the executor. For a processing operation with no result ([Runnable]), we use [runAsync], which returns a [CompletableFuture<Void>];
• Line 12: The method [state()] (Java 19) returns the status of the operation: [RUNNING] (in progress), [SUCCESS] (completed successfully), [FAILED] (terminated due to an exception), or [CANCELLED] (canceled). These are the constants of the [Future.State] enumeration; the last three are final states: a completed operation no longer changes state;
• line 14: [join()] waits for the operation to finish and returns its result. Like [Thread.join] in the previous chapter, this is a blocking wait: the current thread is suspended (the main thread in this case, which is not a problem for a small console program);
• Lines 17–22: A [CompletableFuture] created as "in progress" by its creator and terminated by a timer after 300 ms ([completeOnTimeout], Java 9). No thread is occupied during this time: this is the asynchronous equivalent of [Thread.sleep(300)]. Its type is [CompletableFuture<Void>]: it does not return a result ([Void] is the class corresponding to [void]; its only value is [null]);
• Lines 25–28: [completedFuture(42)] returns a previously completed [CompletableFuture] with a result of 42, and [failedFuture(exception)] (Java 9) returns a [CompletableFuture] that has already failed. They are used when a method needs to return a *[CompletableFuture] but already has the result (a cached value, for example) or already knows that it will fail. [resultNow()] and [exceptionNow()] (Java 19) return the result or exception of a completed operation without waiting (they throw a [IllegalStateException]* if the operation is not in the desired state);
• lines 31–37: An operation whose function throws an exception ends in the [FAILED] state. The exception is stored in [CompletableFuture], and [join()] triggers a [CompletionException] (an unhandled exception) whose cause ([getCause()]) is the original exception, in this case a [NumberFormatException];
• lines 40–45: [get()], the method inherited from the [Future] interface also waits in a blocking state, but wraps the exception in a [ExecutionException]—a checked exception—and can also throw a [InterruptedException], which is also checked: The call is therefore placed within a *[try / catch], which intercepts the first one, and the *[main] method declares the second one (line 2). [join()] is more convenient in code that manipulates [CompletableFuture]; [get()] is the only one available for a simple [Future], and there is a version with a delay, [get(timeout, unit)];
• Lines 48–50: a continuation. [f2.thenApply(function)] returns a new [CompletableFuture], which will be completed with the result of the function applied to the result of [f2] once [f2] is completed. Here, the function ignores its parameter ([null]) and returns 42: this results in a [CompletableFuture<Integer>]. [thenAccept] attaches a function that consumes the result without producing one ([Consumer<T>]) and returns a [CompletableFuture<Void>]. If the original operation fails, the continuation is not executed and the failure propagates along the chain;
• lines 53–55: the Boolean methods [isDone()] (completed, regardless of how), [isCompletedExceptionally()] (completed with an exception or aborted), and [isCancelled()] (aborted).

There are numerous methods in [CompletableFuture] (more than fifty). The following table lists the ones we will use:

Method	Role
[supplyAsync(f)], [runAsync(r)]	Execute a function (with or without a result) using an executor
[completedFuture(v)], [failedFuture(e)]	Create an operation that has already been completed, whether successfully or unsuccessfully
[complete(v)], [completeExceptionally(e)]	complete a task on your own
[thenApply(f)]	Transform the result: [T → U]
[thenAccept(c)], [thenRun(r)]	Use the result, or perform an action after the operation
[thenCompose(f)]	Chain another asynchronous operation: [T → CompletableFuture<U>]
[thenCombine(other, f)]	Combine the results of two operations performed in parallel
[allOf(...)], [anyOf(...)]	Wait for all operations to complete, or for the first one to finish
[exceptionally(f)], [handle(f)], [whenComplete(c)]	Handle or observe a failure
[orTimeout(...)], [completeOnTimeout(...)]	limit the duration of an operation
[join()], [get()]	Wait for the result by blocking the current thread

The methods [thenXxx], [handle], and [whenComplete] also exist in the [thenXxxAsync] version: we’ll discuss the difference in the section “Which thread executes the rest of the code?”

9.3. Asynchronous methods

9.3.1. First approach: returning a CompletableFuture

In Java, an asynchronous method is simply a method that initiates an operation and immediately returns, without waiting, a [CompletableFuture] that represents it:

• [CompletableFuture<Void>] if the operation returns no result;
• [CompletableFuture<T>] if it returns a result of type [T].

The language does not impose any naming conventions. Java libraries often suffix these methods with [Async] ([sendAsync], [supplyAsync]), but this is not a rule: the return type is sufficient to identify them.

The program [Async04.java] makes coffee and counts sugar cubes:

// methods that return a CompletableFuture: the steps are composed with thenRun and thenCompose
final long start = System.nanoTime();

void main() {
    display("call of makeCoffee");
    CompletableFuture<Void> preparation = makeCoffee();
    display("makeCoffee has returned: the program continues");
    display("call of countSugars");
    CompletableFuture<Integer> counting = countSugars(3);
    display("countSugars has returned");
    // we combine the two operations, then wait for the result of the combination
    int nbSugars = preparation.thenCombine(counting, (nothing, count) -> count).join();
    display("coffee ready, with " + nbSugars + " sugar(s)");
}

// an operation that returns no result: it returns a CompletableFuture<Void>
CompletableFuture<Void> makeCoffee() {
    display("  coffee: the machine is heating up");
    // the method describes the sequence of steps, then returns immediately
    return delay(1000)
            .thenRun(() -> display("  coffee: the machine is pouring the coffee"))
            .thenCompose(nothing -> delay(500))
            .thenRun(() -> display("  coffee: done"));
}

// an operation that returns an integer: it returns a CompletableFuture<Integer>
CompletableFuture<Integer> countSugars(int count) {
    // a chain of steps that starts with an already completed CompletableFuture
    CompletableFuture<Void> steps = CompletableFuture.completedFuture(null);
    for (int i = 1; i <= count; i++) {
        // a lambda can only use an effectively final local variable
        int number = i;
        steps = steps.thenCompose(nothing -> delay(200))
                .thenRun(() -> display("  sugar no. " + number));
    }
    // at the end of the chain, the result: the number of sugars
    return steps.thenApply(nothing -> count);
}

// the asynchronous equivalent of Thread.sleep: a CompletableFuture completed after [ms] milliseconds
CompletableFuture<Void> delay(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// display with the time elapsed since the start
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async04.java
    3 ms : call of makeCoffee
   12 ms :   coffee: the machine is heating up
   15 ms : makeCoffee has returned: the program continues
   16 ms : call of countSugars
   17 ms : countSugars has returned
  217 ms :   sugar no. 1
  417 ms :   sugar no. 2
  618 ms :   sugar no. 3
 1014 ms :   coffee: the machine is pouring the coffee
 1515 ms :   coffee: done
 1516 ms : coffee ready, with 3 sugar(s)

Comments:

• Lines 40–43: The [delay] method is the asynchronous equivalent of [Thread.sleep]: it returns a [CompletableFuture<Void>] that will complete after the timeout, without occupying a thread. We will use it in many examples to simulate I/O;
• Lines 17–24: an operation with no result. The method displays an initial message, then describes the sequence of steps and immediately returns the final [CompletableFuture]: this is why line 4 of the results appears right after line 3, rather than a second and a half later. The steps will be executed later, each one once the previous one is complete:
  • line 21: [thenRun(action)] performs an action when the previous operation is complete;
  • line 22: [thenCompose(function)] triggers another asynchronous operation: the function itself returns a [CompletableFuture] (in this case, a new 500-ms delay), and the [CompletableFuture] returned by [thenCompose] will not be completed until this second operation is finished. With [thenApply], we would have obtained a [CompletableFuture<CompletableFuture<Void>>], which would have completed as soon as the second delay began: [thenCompose] is to [thenApply] as [flatMap] is to [map] in the API stream;
• Lines 27–38: An operation that returns an integer. A loop cannot “wait” on every iteration: it builds the sequence of steps, with each iteration adding a delay and a display to the existing sequence, starting from a pre-computed [CompletableFuture] (line 29). The entire string is constructed in a few microseconds and then unfolds as the delays occur. The variable [number] (line 32) is necessary because a lambda can only use a local variable that is effectively final, which [i] is not. Finally, [thenApply] converts the string into a [CompletableFuture<Integer>], the result of which is the number of sugars;
• Lines 6–10: The main program calls both methods. It retrieves two [CompletableFuture] values and continues execution. The two operations then proceed “simultaneously”: the sugars are counted (lines 7–9 of the results) while the machine warms up;
• line 12: [preparation.thenCombine(counting, function)] returns a completed [CompletableFuture] when both operations are finished, along with the result of the function applied to their two results: in this case, the number of sugar cubes is retained. [join()] waits for this result. The coffee is ready after 1.5 seconds—the duration of the longer of the two operations—and not after 2.1 seconds (1.5 + 3 × 0.2).

We can see the strengths and limitations of this approach. No thread waits: the steps are executed by the timer thread as the time intervals elapse. But the code no longer reads like a sequence of instructions: the [countSugars] loop doesn’t do what a loop usually does (it constructs a string instead of executing iterations), and a simple [if] or [try / catch] in the middle of the steps requires dedicated methods. That’s the price you pay when the language doesn’t offer keywords like [await].

9.3.2. Second approach: blocking code on virtual threads

Virtual threads allow you to write the same methods as ordinary, blocking methods. The program [Async05.java] makes the same cup of coffee:

// the same coffee with virtual threads: ordinary, blocking methods that read from top to bottom
final long start = System.nanoTime();

void main() throws InterruptedException, ExecutionException {
    // an executor that creates a virtual thread for each task given to it
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        display("start of makeCoffee");
        Future<?> preparation = executor.submit(this::makeCoffee);
        display("makeCoffee is started: the program continues");
        display("start of countSugars");
        Future<Integer> counting = executor.submit(() -> countSugars(3));
        display("countSugars is started");
        // we now wait for the two results
        preparation.get();
        int nbSugars = counting.get();
        display("coffee ready, with " + nbSugars + " sugar(s)");
    }
}

// an ordinary method that returns no result
void makeCoffee() {
    display("  coffee: the machine is heating up");
    // a blocking wait: on a virtual thread, it releases the carrier thread
    sleep(1000);
    display("  coffee: the machine is pouring the coffee");
    sleep(500);
    display("  coffee: done");
}

// an ordinary method that returns an integer
int countSugars(int count) {
    for (int i = 1; i <= count; i++) {
        sleep(200);
        display("  sugar no. " + i);
    }
    return count;
}

// Thread.sleep without checked exception: an interruption becomes a CancellationException
void sleep(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        // we restore the interrupt flag of the thread before throwing the exception
        Thread.currentThread().interrupt();
        throw new CancellationException("wait interrupted");
    }
}

// display with the time elapsed since the start
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async05.java
    3 ms : start of makeCoffee
   17 ms :   coffee: the machine is heating up
   18 ms : makeCoffee is started: the program continues
   18 ms : start of countSugars
   19 ms : countSugars is started
  221 ms :   sugar no. 1
  422 ms :   sugar no. 2
  623 ms :   sugar no. 3
 1020 ms :   coffee: the machine is pouring the coffee
 1521 ms :   coffee: done
 1522 ms : coffee ready, with 3 sugar(s)

Comments:

• Lines 20–37: Both operations are regular methods: [makeCoffee] returns nothing ([void]), and [countSugars] returns a [int]. The calls are blocking [Thread.sleep] operations; the loop is a true loop. Nothing in these methods indicates that they will be executed concurrently: they could be called directly;
• lines 39–48: the method [sleep] calls [Thread.sleep] and converts any [InterruptedException] (checked exception) into a [CancellationException] (unchecked exception), after resetting the thread’s interrupt flag: this is the rule when intercepting a [InterruptedException] without propagating it (we’ll return to this in the section on cancellation);
• line 6: an executor that creates a virtual thread for each task. It is declared in a [try] with a resource: its [close()] method will wait for all tasks to complete;
• Lines 8 and 11: [submit] delegates each method to a new virtual thread and returns a [Future]. The method reference [this::makeCoffee] is a [Runnable] (the method returns nothing), resulting in a [Future<?>]; the lambda [() -> countSugars(3)] is a [Callable<Integer>], resulting in a [Future<Integer>];
• Lines 14–15: The main program waits for both results using [get()]. This is a blocking wait by the main thread, which has no consequence here;
• the results are the same as with [Async04.java]: 1.5 seconds total, with the sugar counted while the machine warms up. During the waits, no system threads are blocked: the two virtual threads are detached from their carrier threads.

With virtual threads, asynchrony is no longer a matter of code writing but of execution: we write blocking code, and it is the executor’s choice (virtual threads rather than a platform thread pool) that determines whether the waits block system threads. This is why existing Java libraries (JDBC, files, sockets, [HttpClient.send]…), all written in a blocking manner, take advantage of virtual threads without modification.

9.3.3. What JVM Does

How can a virtual thread be interrupted in the middle of a loop, release its host thread, and then resume later at the same point, with the same local variable values, possibly on a different host thread? A virtual thread is a Java object ([java.lang.VirtualThread], a class derived from [Thread]) that has its own call stack. To execute it, JVM “mounts” it onto a carrier thread—a platform thread of a [ForkJoinPool] reserved for this purpose, which by default has as many threads as there are processors. When the virtual thread calls a blocking operation ([Thread.sleep], reading from a socket; [BlockingQueue.take], waiting for a lock…), the classes in JVM—rewritten for this purpose in Java 21—do not block the containing thread:

• they register the operation with the system (a timer for a delay, the system’s I/O notification mechanism—[epoll] on Linux—for a socket);
• they “unmount” the virtual thread: its call stack frames are copied to the heap as an object called a continuation, and the carrier thread is returned to [ForkJoinPool], which assigns it another virtual thread ready to execute;
• when the operation completes, the virtual thread becomes ready again: it will be remounted onto the first available carrier thread, its stack frames will be recopied, and execution will resume immediately after the blocking call.

This is the principle behind callback methods (“call me back when the operation is complete”), but it is the JVM that breaks down the execution, saves the variables, and restores the context at the call stack level, without the source code showing any trace of it. Other languages entrust this task to the compiler, which transforms each [async] method into a “state machine”; Java entrusts it to the JVM, which allows it to handle all existing code in the same way.

There are two limitations to be aware of:

• certain operations cannot detach the virtual thread; it therefore remains “pinned” to its host thread, which is blocked while waiting. This occurs during the execution of a native method, and during most file reads and writes (the system does not provide a notification for files that can be used by the JVM); for the latter, the [ForkJoinPool] compensates by temporarily creating an additional carrier thread. Up to Java 23, this was also the case for a virtual thread waiting inside a [synchronized] block; since Java 24, this is no longer the case;
• a virtual thread does not speed up a computation: a virtual thread performing a computation occupies its host thread for the entire duration of the computation. Virtual threads are intended for programs that do a lot of waiting, not those that do a lot of computation (for the latter, thread pools and parallel streams from the previous chapter are available).

9.3.4. Which thread executes the next section?

The program [Async06.java] displays the thread that executes each piece of code:

// which thread executes the rest of a processing?
void main() throws InterruptedException {
    display("start of the program, parallelism of the common pool: " + ForkJoinPool.getCommonPoolParallelism());

    // 1 - the continuation of an already completed CompletableFuture is executed immediately, by the caller
    CompletableFuture.completedFuture(42).thenAccept(value -> display("1 - thenAccept, result known"));

    // 2 - otherwise, it is executed by the thread that completes the CompletableFuture
    delay(100).thenRun(() -> display("2 - thenRun after a delay")).join();
    CompletableFuture<String> order = new CompletableFuture<>();
    order.thenAccept(text -> display("2 - thenAccept after complete(\"" + text + "\")"));
    Thread.ofPlatform().name("deliverer").start(() -> order.complete("delivered")).join();

    // 3 - the xxxAsync variants give the continuation to an executor
    delay(100).thenRunAsync(() -> display("3 - thenRunAsync, default executor")).join();
    ThreadFactory factory = Thread.ofPlatform().name("pool-", 1).factory();
    try (ExecutorService pool = Executors.newFixedThreadPool(2, factory)) {
        delay(100).thenRunAsync(() -> display("3 - thenRunAsync, chosen executor"), pool).join();
        CompletableFuture.runAsync(() -> display("3 - runAsync, chosen executor"), pool).join();
    }
    try (ExecutorService virtuals = Executors.newVirtualThreadPerTaskExecutor()) {
        CompletableFuture.runAsync(() -> display("3 - runAsync, virtual threads"), virtuals).join();
    }

    // 4 - a virtual thread may change carrier thread after each wait
    Thread a = Thread.ofVirtual().start(() -> work("A"));
    Thread b = Thread.ofVirtual().start(() -> work("B"));
    a.join();
    b.join();
    display("end of the program");
}

void work(String name) {
    for (int i = 1; i <= 3; i++) {
        display("  " + name + i + " : before the wait");
        sleep(100);
    }
}

// a CompletableFuture completed after [ms] milliseconds
CompletableFuture<Void> delay(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// Thread.sleep without checked exception: an interruption becomes a CancellationException
void sleep(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("wait interrupted");
    }
}

// display prefixed by the thread executing it
// (name of a platform thread, description of a virtual thread)
void display(String message) {
    Thread thread = Thread.currentThread();
    IO.println("[" + (thread.isVirtual() ? thread.toString() : thread.getName()) + "] " + message);
}

Results:

> java Async06.java
[main] start of the program, parallelism of the common pool: 1
[main] 1 - thenAccept, result known
[ForkJoinPool.commonPool-delayScheduler] 2 - thenRun after a delay
[deliverer] 2 - thenAccept after complete("delivered")
[ForkJoinPool.commonPool-worker-1] 3 - thenRunAsync, default executor
[pool-1] 3 - thenRunAsync, chosen executor
[pool-2] 3 - runAsync, chosen executor
[VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1] 3 - runAsync, virtual threads
[VirtualThread[#31]/runnable@ForkJoinPool-1-worker-1]   A1 : before the wait
[VirtualThread[#32]/runnable@ForkJoinPool-1-worker-2]   B1 : before the wait
[VirtualThread[#32]/runnable@ForkJoinPool-1-worker-2]   B2 : before the wait
[VirtualThread[#31]/runnable@ForkJoinPool-1-worker-1]   A2 : before the wait
[VirtualThread[#31]/runnable@ForkJoinPool-1-worker-2]   A3 : before the wait
[VirtualThread[#32]/runnable@ForkJoinPool-1-worker-1]   B3 : before the wait
[main] end of the program

Comments:

• line 3: the shared pool ([ForkJoinPool.commonPool()]), the default executor for [CompletableFuture], has a parallelism equal to the number of processors minus one, which is 1 on our 2-processor machine (line 2 of the results);
• Line 6: The continuation of a [CompletableFuture] that has already completed is executed immediately by the thread that attached it: thread [main] (line 3);
• lines 9–12: if the operation is not complete, the continuation will be executed by the thread that will complete it: the timeout timer thread for [delay(100)] (line 4), the thread [deliverer], which calls [complete] for the order (line 5). Therefore, the thread that executes a non-asynchronous continuation is not chosen; this is irrelevant for a brief action, but a long or blocking continuation would immobilize that thread (the timer thread, for example, which handles all the timeouts for [CompletableFuture]);
• lines 15–23: the [xxxAsync] variants ([thenRunAsync], [thenApplyAsync]…) delegate the continuation to an executor: without additional parameters, to the shared pool (line 6 of the results); with an executor as the last parameter, to that executor. This allows one to choose platform threads (lines 7–8: a pool of two threads whose names are determined by a thread factory, [ThreadFactory], constructed by [Thread.ofPlatform().name("pool-", 1).factory()]), or virtual threads (line 9). [supplyAsync] and [runAsync] also accept an executor;
• Lines 26–38: Two virtual threads that display a message and then sleep, three times in a row. The results show that they do not remain attached to a single host thread: virtual thread No. 31 (A) starts at [worker-1] and ends at [worker-2], while No. 32 (B) does the opposite. While one is sleeping, its carrier thread is free for the other. Therefore, in a virtual thread, one must never assume that the code following a wait is executed by the same carrier thread as the code preceding it; but it is always the same virtual thread, and that is what matters to the program ([Thread.currentThread()] returns the virtual thread; its variables, [ThreadLocal], are preserved).

This method of performing multiple operations simultaneously without increasing the number of system threads is called concurrency. It differs from parallelism, which involves executing multiple calculations at the same time across multiple processors (as discussed in the previous chapter: thread pools, parallel streams). Asynchrony facilitates I/O concurrency; it does not make calculations faster.

The choice of executor is therefore important:

• the shared pool is suitable for brief continuations and short computations; blocking operations should not be executed there, as this risks the starvation described in [Async01.java] (it has only one thread on our machine);
• for a blocking operation (I/O, calls to JDBC or [Thread.sleep]), use a virtual thread executor: [CompletableFuture.supplyAsync(supplier, virtualExecutor)];
• for a graphical user interface, we pass an executor that has the continuation executed by the UI thread, which allows the graphical components to be updated: with Swing, [thenAcceptAsync(action, SwingUtilities::invokeLater)] (the method reference [SwingUtilities::invokeLater] is a valid [Executor], since [Executor] is a functional interface whose sole method takes a [Runnable]); with JavaFX, [Platform::runLater].

9.3.5. The Main Program

In other languages, the main method of an asynchronous program must be declared in a specific way. In Java, there is no such requirement: the method [main] is executed by the platform thread [main], and it is up to that method to wait for the operations it has initiated—using [join()] or [get()]—to complete. This is the only place where a blocking wait on a [CompletableFuture] is normal. The [Async07.java] program performs an asynchronous countdown:

// the main program waits for the end of the asynchronous operations and sets the exit code
void main(String[] args) {
    // the number of seconds is passed as an argument
    if (args.length != 1 || !args[0].matches("\\d+")) {
        IO.println("syntax: java Async07.java nbSeconds");
        // the exit code of the program: 0 for a success, something else for an error
        System.exit(1);
    }
    CompletableFuture<Void> countdown = countDown(Integer.parseInt(args[0]));
    IO.println("countdown started");
    // without this wait, main would end immediately and the program with it
    countdown.join();
    IO.println("finished");
}

// an asynchronous loop is written with recursion: each step starts the next one
CompletableFuture<Void> countDown(int nbSeconds) {
    if (nbSeconds == 0) {
        return CompletableFuture.completedFuture(null);
    }
    IO.println("still " + nbSeconds + " s");
    return delay(1000).thenCompose(nothing -> countDown(nbSeconds - 1));
}

// a CompletableFuture completed after [ms] milliseconds
CompletableFuture<Void> delay(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

Results, first with and then without an argument:

> java Async07.java 3
still 3 s
countdown started
still 2 s
still 1 s
finished

> java Async07.java
syntax: java Async07.java nbSeconds
> echo $?
1

Comments:

• line 2: the [args] parameter of [main] receives the command-line arguments, which are placed after the filename: [java Async07.java 3];
• Lines 4–8: If the argument is missing or is not a positive integer (regular expression [\d+]: one or more digits), the program terminates with [System.exit(1)]. The integer passed to [System.exit] is the program’s exit code (0 for success, any other value for an error), which can be checked by a script: [echo $?] on Linux (second result), [echo %ERRORLEVEL%] on Windows. A program whose [main] method terminates normally returns code 0;
• lines 16–23: an asynchronous countdown. An asynchronous loop is naturally implemented using recursion: each step displays a message and then, using [thenCompose], introduces a one-second delay followed by the next step. There is no call stack: the recursive call is made via the continuation, one second later;
• lines 9–12: the method [countDown] displays its first message and then returns control (lines 2–3 of the results); the main program waits for the operation to finish with [join()]. Without this wait, [main] would terminate immediately, and the program along with it: JVM stops when only daemon threads remain, and the timer and shared pool threads are daemons. The same applies to virtual threads, which are always daemons: a program must wait for the virtual threads it has launched to finish (using [join], [get], or the [close] method of their executor).

9.4. Sequential Execution and Concurrent Execution

9.4.1. allOf and invokeAll

The coffee example demonstrated this: the moment an asynchronous operation is initiated and the moment it is awaited can be different. Writing [method().join()] initiates the operation and awaits it immediately; writing [var f = method();] and then, later, [f.join();] allows other operations to be launched in between. The program [Async08.java] compares the two approaches using operations that take one second:

// sequential execution and concurrent execution of one-second operations
long start = System.nanoTime();

void main() throws InterruptedException {
    // 1 - in sequence: each operation is awaited before starting the next one
    int a = readTemperature("Angers").join();
    int b = readTemperature("Brest").join();
    int c = readTemperature("Colmar").join();
    IO.println("in sequence: " + a + ", " + b + ", " + c + " in " + elapsed() + " ms");

    // 2 - concurrently: we start the three operations, then we wait for them
    start = System.nanoTime();
    CompletableFuture<Integer> fa = readTemperature("Angers");
    CompletableFuture<Integer> fb = readTemperature("Brest");
    CompletableFuture<Integer> fc = readTemperature("Colmar");
    CompletableFuture.allOf(fa, fb, fc).join();
    IO.println("concurrently: " + fa.resultNow() + ", " + fb.resultNow() + ", " + fc.resultNow()
            + " in " + elapsed() + " ms");

    // 3 - allOf on a list of CompletableFutures, then the list of the results
    start = System.nanoTime();
    List<String> cities = List.of("Angers", "Brest", "Colmar", "Dijon", "Évry", "Foix");
    List<CompletableFuture<Integer>> readings = cities.stream().map(this::readTemperature).toList();
    List<Integer> temperatures = CompletableFuture.allOf(readings.toArray(CompletableFuture[]::new))
            .thenApply(nothing -> readings.stream().map(CompletableFuture::join).toList())
            .join();
    IO.println(cities.size() + " cities: " + temperatures + " in " + elapsed() + " ms");

    // 4 - the same thing with virtual threads and a blocking method
    start = System.nanoTime();
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        List<Callable<Integer>> tasks = cities.stream()
                .map(city -> (Callable<Integer>) () -> readTemperatureBlocking(city))
                .toList();
        // invokeAll starts all the tasks and returns their Futures when they are all finished
        List<Integer> results = executor.invokeAll(tasks).stream().map(Future::resultNow).toList();
        IO.println(cities.size() + " cities (virtual threads): " + results + " in " + elapsed() + " ms");
    }
}

// simulates querying a remote service that answers in one second
CompletableFuture<Integer> readTemperature(String city) {
    // a "temperature" derived from the name of the city
    return delay(1000).thenApply(nothing -> 10 + city.length());
}

// the same operation, written in a blocking way
int readTemperatureBlocking(String city) throws InterruptedException {
    Thread.sleep(1000);
    return 10 + city.length();
}

// a CompletableFuture completed after [ms] milliseconds
CompletableFuture<Void> delay(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// milliseconds elapsed since [start]
long elapsed() {
    return (System.nanoTime() - start) / 1_000_000;
}

Results:

> java Async08.java
in sequence: 16, 15, 16 in 3007 ms
concurrently: 16, 15, 16 in 1000 ms
6 cities: [16, 15, 16, 15, 14, 14] in 1006 ms
6 cities (virtual threads): [16, 15, 16, 15, 14, 14] in 1013 ms

Comments:

• lines 41–45: the simulated operation: a one-second delay, followed by a “temperature” derived from the length of the city name;
• lines 6–8: Each operation is started and then waited for before starting the next one. The three operations are performed one after the other: 3 seconds (line 2 of the results). This is the correct approach when one operation requires the result of the previous one (in which case, one would instead write a chain of [thenCompose]);
• lines 13–15: The three operations are started without waiting for each other. They run simultaneously;
• line 16: [CompletableFuture.allOf(...)] returns a [CompletableFuture<Void>], which completes once all the operations passed to it are finished. We wait for it with [join()]: 1 second instead of 3 (line 3);
• lines 17–18: The three operations are complete, so [resultNow()] returns their results immediately;
• lines 22–27: the case of a list of operations. [allOf] waits for an array ([CompletableFuture<?>...]): the [toArray(CompletableFuture[]::new)] method in the list creates it. [allOf] does not return the results: they are collected in a continuation by calling [join()] on each operation, which does not block since they have all completed. The list of results is obtained in the order of the operations (not in the order in which they completed). The expression [cities.stream().map(this::readTemperature).toList()] calls the method for each city, thereby launching six operations simultaneously; all six together take one second (line 4);
• lines 29–38: the same thing with a blocking method (lines 47–51) executed by virtual threads. The method [invokeAll] of an executor receives a collection of [Callable<T>], launches them all, and returns the list of their [Future<T>] when they are all complete: This is the blocking equivalent of *[allOf]. The type cast [(Callable<Integer>)] on line 33 tells the compiler the type of the lambda, which it cannot infer on its own in a [map]*. The six operations together take one second (line 5).

9.4.2. anyOf

[CompletableFuture.anyOf(...)] returns a [CompletableFuture<Object>] that terminates as soon as one of the operations passed to it is completed, with the result (or exception) of that operation. The program [Async09.java] uses it in two ways:

// anyOf: reacting to the first completed operation
long start = System.nanoTime();

void main() {
    // 1 - we query three mirror servers and keep the first answer
    List<CompletableFuture<String>> requests = List.of(
            query("mirror-paris", 800),
            query("mirror-lyon", 300),
            query("mirror-lille", 600));
    // anyOf returns a CompletableFuture<Object> that completes like the first completed operation
    Object first = CompletableFuture.anyOf(requests.toArray(CompletableFuture[]::new)).join();
    display("first answer: " + first);
    // the other requests continue: we wait for them here to see their answers
    CompletableFuture.allOf(requests.toArray(CompletableFuture[]::new)).join();
    display("all the answers: " + requests.stream().map(CompletableFuture::resultNow).toList());

    // 2 - a time-limited operation: we "race" it against a delay
    start = System.nanoTime();
    CompletableFuture<String> slow = query("slow-server", 2000);
    CompletableFuture<Object> race = CompletableFuture.anyOf(slow, delay(500));
    race.join();
    if (slow.isDone()) {
        display("answer obtained in time: " + slow.join());
    } else {
        display("the slow server did not answer within the 500 ms delay");
    }
}

// simulates a server that answers after [duration] milliseconds
CompletableFuture<String> query(String server, int duration) {
    return delay(duration).thenApply(nothing -> server + " (" + duration + " ms)");
}

// a CompletableFuture completed after [ms] milliseconds
CompletableFuture<Void> delay(long ms) {
    return new CompletableFuture<Void>().completeOnTimeout(null, ms, TimeUnit.MILLISECONDS);
}

// display with the time elapsed since [start]
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async09.java
  309 ms : first answer: mirror-lyon (300 ms)
  806 ms : all the answers: [mirror-paris (800 ms), mirror-lyon (300 ms), mirror-lille (600 ms)]
  500 ms : the slow server did not answer within the 500 ms delay

Comments:

• lines 6–9: Three mirror servers are queried; they all hold the same information and respond at varying speeds. The three operations are stored in a list;
• line 11: [anyOf] returns a [CompletableFuture<Object>], whose result is that of the first completed operation. Its type is [Object] because the operations passed to [anyOf] may have different result types: it would need to be cast to use it as a string. The first response arrives after 300 ms (line 2 of the results);
• line 14: the other requests are not terminated by [anyOf]; they continue until they finish. Here, we wait for them to display all the responses (line 3);
• lines 19–26: a common use of [anyOf]: limiting the duration of a wait by setting up a “race” between the operation and a timeout. If the operation is not finished when the race is over, the timeout has won: the operation took too long (line 4). [CompletableFuture] offers a more direct way to write this case, the [orTimeout] method, which is presented later.

9.4.3. Processing Results as They Come In

We often want to process the results of multiple operations as they complete, without waiting for the slowest one. With [CompletableFuture], this is straightforward: simply attach a continuation to each operation. For tasks submitted to an executor, the [ExecutorCompletionService] class (Java 5) returns the [Future] objects in the order in which the tasks complete. The program [Async10.java] processes five “files,” each of which takes between 100 and 900 ms to process, in both ways:

// processing the results in the order in which the operations finish
long start = System.nanoTime();
final int[] durations = {900, 300, 700, 100, 500};

void main() throws InterruptedException, ExecutionException {
    // 1 - with CompletableFutures: one continuation per operation, executed as soon as it finishes
    List<CompletableFuture<Void>> displays = new ArrayList<>();
    for (int i = 0; i < durations.length; i++) {
        displays.add(process("file" + i + ".txt", durations[i]).thenAccept(this::display));
    }
    CompletableFuture.allOf(displays.toArray(CompletableFuture[]::new)).join();
    display("1 - all the files have been processed");

    // 2 - with virtual threads: a CompletionService delivers the Futures in the order they finish
    start = System.nanoTime();
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        CompletionService<String> service = new ExecutorCompletionService<>(executor);
        for (int i = 0; i < durations.length; i++) {
            String file = "file" + i + ".txt";
            int duration = durations[i];
            service.submit(() -> processBlocking(file, duration));
        }
        // take() waits (blocking) for the next finished task
        for (int i = 0; i < durations.length; i++) {
            display(service.take().get());
        }
    }
    display("2 - all the files have been processed");
}

// simulates the processing of a file that lasts [duration] milliseconds
CompletableFuture<String> process(String file, int duration) {
    return new CompletableFuture<String>()
            .completeOnTimeout(file + " processed in " + duration + " ms", duration, TimeUnit.MILLISECONDS);
}

// the same processing, written in a blocking way
String processBlocking(String file, int duration) throws InterruptedException {
    Thread.sleep(duration);
    return file + " processed in " + duration + " ms";
}

// display with the time elapsed since [start]
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async10.java
  107 ms : file3.txt processed in 100 ms
  318 ms : file1.txt processed in 300 ms
  508 ms : file4.txt processed in 500 ms
  707 ms : file2.txt processed in 700 ms
  906 ms : file0.txt processed in 900 ms
  907 ms : 1 - all the files have been processed
  115 ms : file3.txt processed in 100 ms
  314 ms : file1.txt processed in 300 ms
  514 ms : file4.txt processed in 500 ms
  714 ms : file2.txt processed in 700 ms
  914 ms : file0.txt processed in 900 ms
  915 ms : 2 - all the files have been processed

Comments:

• Lines 7–11: Each processing task receives its own continuation, [thenAccept(this::display)], which is executed as soon as it finishes: file 3 (100 ms) is displayed first, file 0 (900 ms) last (lines 2–6 of the results). The method [process] (lines 31–35) returns a completed [CompletableFuture<String>] after the specified duration with the desired message ([completeOnTimeout] directly receives the result value). The list of continuations is used to wait for all displays to finish with [allOf];
• lines 16–27: a [CompletionService<T>] wraps an executor (in this case, a virtual thread executor). Its method [submit] launches a task; its method [take()] waits, blocking, for the next completed task and returns its [Future], whose result is immediately returned by [get()]. The loop in lines 24–26 therefore processes the files in the order in which they are completed (lines 8–12);
• in both cases, the entire process takes 900 ms—the duration of the longest processing step—and each result is displayed as soon as possible.

9.5. Exceptions in Asynchronous Code

9.5.1. Propagation, exceptionally, handle, whenComplete

An exception thrown by an asynchronous operation is not thrown back to the code that initiated the operation: it is stored in [CompletableFuture], which transitions to the [FAILED] state. It then propagates along the continuation chain until it reaches a method that handles it, or until it reaches [join()], which rethrows it wrapped in a [CompletionException]. The program [Async11.java] examines the various cases:

// exceptions in asynchronous code
void main() {
    // 1 - join() throws a CompletionException that wraps the exception of the operation
    try {
        int value = readValue("twelve").join();
        IO.println("value = " + value);
    } catch (CompletionException e) {
        IO.println("1 - join() throws " + name(e) + ", with cause " + name(e.getCause())
                + " : " + e.getCause().getMessage());
    }

    // 2 - exceptionally provides a replacement value in case of failure
    int value = readValue("thirteen").exceptionally(e -> -1).join();
    IO.println("2 - exceptionally: replacement value " + value);

    // 3 - handle receives the result or the exception; whenComplete observes them without changing anything
    for (String text : List.of("14", "fifteen")) {
        String outcome = readValue(text)
                .whenComplete((v, e) -> IO.println("3 - whenComplete saw: " + (e == null ? v : name(e))))
                .handle((v, e) -> e == null ? "value " + v : "failure, " + name(cause(e)))
                .join();
        IO.println("3 - handle returned: " + outcome);
    }

    // 4 - a method that returns a CompletableFuture can also throw an exception right at the call
    try {
        readValue(" ");
        IO.println("4 - the call threw nothing");
    } catch (IllegalArgumentException e) {
        IO.println("4 - it is the call itself that throws " + name(e) + " : " + e.getMessage());
    }

    // 5 - allOf with several failing operations
    List<CompletableFuture<Integer>> readings = Stream.of("1", "two", "3", "four")
            .map(this::readValue)
            .toList();
    try {
        CompletableFuture.allOf(readings.toArray(CompletableFuture[]::new)).join();
    } catch (CompletionException e) {
        // join() rethrows only one of the exceptions…
        IO.println("5 - allOf fails with " + name(e.getCause()) + " : " + e.getCause().getMessage());
        // … but each operation has kept its exception or its result
        for (CompletableFuture<Integer> reading : readings) {
            switch (reading.state()) {
                case SUCCESS -> IO.println("      success, result = " + reading.resultNow());
                case FAILED -> IO.println("      failure, " + name(reading.exceptionNow())
                        + " : " + reading.exceptionNow().getMessage());
                default -> IO.println("      unexpected state: " + reading.state());
            }
        }
    }
}

// an asynchronous operation that fails if the string is not an integer
CompletableFuture<Integer> readValue(String text) {
    // immediate check of the parameter, before starting the operation
    if (text == null || text.isBlank()) {
        throw new IllegalArgumentException("the text must not be empty");
    }
    // the reading takes a little time
    return new CompletableFuture<Void>()
            .completeOnTimeout(null, 100, TimeUnit.MILLISECONDS)
            .thenApply(nothing -> Integer.parseInt(text));
}

// continuations often receive a CompletionException: we go back to its cause
Throwable cause(Throwable e) {
    return e instanceof CompletionException && e.getCause() != null ? e.getCause() : e;
}

// the simple name of the class of an exception
String name(Throwable e) {
    return e.getClass().getSimpleName();
}

Results:

> java Async11.java
1 - join() throws CompletionException, with cause NumberFormatException : For input string: "twelve"
2 - exceptionally: replacement value -1
3 - whenComplete saw: 14
3 - handle returned: value 14
3 - whenComplete saw: CompletionException
3 - handle returned: failure, NumberFormatException
4 - it is the call itself that throws IllegalArgumentException : the text must not be empty
5 - allOf fails with NumberFormatException : For input string: "two"
      success, result = 1
      failure, NumberFormatException : For input string: "two"
      success, result = 3
      failure, NumberFormatException : For input string: "four"

Comments:

• Lines 54–64: the method being tested. It first checks its parameter and launches a [IllegalArgumentException] if the string is [null] or empty. Otherwise, it returns an operation that waits 100 ms and then converts the entire string, which throws a [NumberFormatException] if the string is not a number;
• lines 4–10: the standard case. The exception thrown by [Integer.parseInt] in the continuation of line 63 is caught by [CompletableFuture]; [join()] rethrows the exception wrapped in a [CompletionException], from which [getCause()] returns the original exception (line 2 of the results);
• line 13: [exceptionally(function)] returns a [CompletableFuture] that has the same result as the operation if it succeeds, and otherwise the result of the function applied to the exception: this is the [catch] from the world of [CompletableFuture]. Here, a failure is replaced with the value -1 (line 3);
• lines 17–23: two other methods, applied to a successful operation and then to a failed operation:
  • [whenComplete(action)] receives the result and the exception (one of the two is [null]) and does not change anything: the [CompletableFuture] it returns has the same result or the same exception as the operation. It is used for observation (logging, freeing a resource): this is the [finally] of the [CompletableFuture] world;
  • [handle(function)] also receives the result and the exception, and returns a new value in both cases: it transforms both a success and a failure into a result;
  • Lines 6 and 7 of the results reveal a pitfall: the exception received by the continuations is not [NumberFormatException] but a [CompletionException] that wraps it, because the failure occurred in a previous step of the chain (the continuation on line 63). The method [cause] (lines 66–69) traces back to the original exception: we’ll get into the habit of writing it out;
• lines 26–31: an important difference from the [async] methods in other languages: a method that returns a [CompletableFuture] is an ordinary method, which can throw an exception as soon as it is called, before creating any [CompletableFuture]. This is the case here for an empty string (line 8 of the results). The calling code must therefore expect both types of errors, or the method must return an operation that has already failed ([return CompletableFuture.failedFuture(new IllegalArgumentException(...));]), which is often preferable;
• lines 34–38: We wait with [allOf] for four operations, two of which fail. The [CompletableFuture] from [allOf] fails in turn, but with only one of the exceptions (line 9 of the results);
• lines 43–50: Each operation has retained its result or exception. We examine their status using a [switch]; [exceptionNow()] returns the original exception, without a wrapper. The failure of [allOf] does not prevent the other operations from succeeding (lines 10 and 12).

With virtual threads, the issue does not arise in the same way: the code executed by a virtual thread is ordinary code, where exceptions propagate as usual and are intercepted by [try / catch]. Only the boundary between two threads requires special attention: an exception from a task submitted to an executor is stored in its [Future], and [get()] rethrows it wrapped in a [ExecutionException].

9.5.2. Forgotten Operations

What happens to the exception from an operation that no one is waiting for? The program [Async12.java] demonstrates this:

// an operation started then forgotten: its exception is lost
void main() {
    // the handler of the exceptions that no code has caught in a thread
    Thread.setDefaultUncaughtExceptionHandler((thread, e) ->
            IO.println("uncaught exception in [" + thread.getName() + "] : " + e.getMessage()));

    // 1 - an awaited CompletableFuture: its exception is caught
    try {
        save("a").join();
    } catch (CompletionException e) {
        IO.println("1 - awaited CompletableFuture: exception caught: " + e.getCause().getMessage());
    }

    // 2 - a forgotten CompletableFuture: its exception stays in the object, nobody will see it
    save("b");
    IO.println("2 - forgotten CompletableFuture: the call has returned");

    // 3 - a task submitted to an executor and forgotten: its exception stays in the Future
    ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
    executor.submit(() -> saveBlocking("c"));
    IO.println("3 - forgotten Future: the task is started");

    // 4 - a thread: the uncaught exception ends the thread and reaches the handler
    Thread.ofVirtual().name("recorder").start(() -> saveBlocking("d"));
    IO.println("4 - virtual thread started");

    // 5 - start without waiting, but handling the failure
    save("e").exceptionally(e -> {
        IO.println("5 - failure reported by exceptionally: " + e.getCause().getMessage());
        return null;
    });

    // the program continues its work…
    for (int i = 1; i <= 4; i++) {
        sleep(100);
        IO.println("the program is working (" + i + ")");
    }
    executor.close();
    IO.println("normal end of the program");
}

// an asynchronous operation that fails after 200 ms
CompletableFuture<Void> save(String data) {
    return new CompletableFuture<Void>()
            .completeOnTimeout(null, 200, TimeUnit.MILLISECONDS)
            .thenRun(() -> {
                throw new IllegalStateException("failed to save [" + data + "]");
            });
}

// the same operation, written in a blocking way
void saveBlocking(String data) {
    sleep(200);
    throw new IllegalStateException("failed to save [" + data + "]");
}

// Thread.sleep without checked exception: an interruption becomes a CancellationException
void sleep(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("wait interrupted");
    }
}

Results:

> java Async12.java
1 - awaited CompletableFuture: exception caught: failed to save [a]
2 - forgotten CompletableFuture: the call has returned
3 - forgotten Future: the task is started
4 - virtual thread started
the program is working (1)
5 - failure reported by exceptionally: failed to save [e]
uncaught exception in [recorder] : failed to save [d]
the program is working (2)
the program is working (3)
the program is working (4)
normal end of the program

Comments:

• lines 4–5: [Thread.setDefaultUncaughtExceptionHandler] sets up the handler that is called when an exception is not caught by any code in a thread, just before the end of that thread. Without it, JVM displays the exception’s stack trace in the error output;
• lines 8–12: The operation is expected, and its exception is intercepted normally;
• lines 14–16: the operation is launched and then forgotten. Its exception is stored in [CompletableFuture], which no one will ever consult: it is lost, without so much as a message;
• lines 18–21: The same applies to a task submitted by [submit] to an executor: its exception is stored in the returned [Future], which is ignored;
• line 24: a thread launched directly has no [Future]: the exception it fails to handle terminates the thread and reaches the uncaught exception handler (line 8 of the results). The program, however, continues;
• lines 28–31: the correct way to start an operation without waiting for it: attach a continuation [exceptionally] (or [whenComplete]) to it that handles failure, in this case by printing a message (line 7);
• Lines 33–39 show that, in all cases, the program continues running until its normal termination. This differs from some languages, where an uncaught exception can terminate the program; in Java, it goes unnoticed, which isn’t any better: a record error that no one sees means lost data.

The rule is therefore: an operation that has been initiated must be completed ([join], [get], [allOf], or the [close] method of an executor, which waits for its tasks), or receive a continuation that handles its failure. Code analysis tools (SpotBugs, IntelliJ IDEA's inspections) may also flag the ignored result of a method that returns a [Future].

9.6. Cancellation and Deadlines

9.6.1. Future.cancel and thread termination

A long-running operation must be able to be canceled: the user clicked on [Cancel], a server client disconnected, the application shut down, or a timeout occurred. Java uses a cooperative mechanism for this—thread interruption:

• a thread’s [interrupt()] method sets its interrupt flag; it does not stop the thread (the [Thread.stop] method, which forced a stop, has been deprecated since Java 20: it calls [UnsupportedOperationException]);
• if the thread is blocked in an interruptible wait ([Thread.sleep], [Object.wait], [Thread.join], [BlockingQueue.take], [Future.get]…), the wait is immediately interrupted and triggers a [InterruptedException], clearing the flag;
• a thread that performs calculations checks its own indicator regularly using [Thread.currentThread().isInterrupted()] and terminates gracefully if it is set;
• the [cancel(true)] method of a [Future] returned by an executor transitions the task to the [CANCELLED] state and interrupts the thread executing it ([cancel(false)] simply prevents the task from starting if it has not yet done so). After cancellation, [get()] launches a [CancellationException].

The mechanism is cooperative: no one forcibly stops an operation; the operation stops itself at a point where it can do so cleanly. This is why a method that intercepts a [InterruptedException] must not ignore it: it must either propagate it, terminate itself, or at the very least reset the flag with [Thread.currentThread().interrupt()] so that the calling code knows that an interruption has been requested (this is what our method [sleep] does).

The program [Async13.java] cancels a simulated download in several ways:

// cancelling an operation: Future.cancel(true) and thread interruption
long start = System.nanoTime();

void main() throws InterruptedException, ExecutionException {
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        // 1 - we do not wait for the result more than 1.2 seconds, then we cancel the operation
        Future<?> download1 = executor.submit(() -> download(10));
        try {
            download1.get(1200, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            // cancel(true) interrupts the thread executing the task
            download1.cancel(true);
            display("1 - " + e.getClass().getSimpleName() + ", state: " + download1.state());
        }

        // 2 - a computation that checks itself, regularly, whether it has been interrupted
        start = System.nanoTime();
        Future<Long> computation = executor.submit(this::compute);
        Thread.sleep(300);
        computation.cancel(true);
        try {
            display("2 - result: " + computation.get());
        } catch (CancellationException e) {
            display("2 - get() throws " + e.getClass().getSimpleName() + ", state: " + computation.state());
        }
        Thread.sleep(100);

        // 3 - the cancellation is requested by the program after 700 ms
        start = System.nanoTime();
        Future<?> download3 = executor.submit(() -> download(10));
        Thread.sleep(700);
        display("3 - the program requests the cancellation");
        download3.cancel(true);
        display("3 - isCancelled=" + download3.isCancelled());
        Thread.sleep(100);

        // 4 - cancel() on a CompletableFuture does not interrupt the operation it represents
        start = System.nanoTime();
        CompletableFuture<Void> download4 = CompletableFuture.runAsync(() -> download(3), executor);
        Thread.sleep(400);
        download4.cancel(true);
        display("4 - CompletableFuture cancelled: isCancelled=" + download4.isCancelled()
                + ", but the operation continues");
    } // close() waits for the end of all the tasks
}

// simulates the download of [nbBlocks] blocks of data (one block every 300 ms)
void download(int nbBlocks) {
    try {
        for (int i = 1; i <= nbBlocks; i++) {
            // a blocking wait is interrupted as soon as the thread is interrupted
            Thread.sleep(300);
            display("  block " + i + "/" + nbBlocks + " received");
        }
    } catch (InterruptedException e) {
        display("  download interrupted");
    }
}

// a long computation that checks the interrupt flag of its thread every 1,000,000 rounds
long compute() {
    long sum = 0;
    for (long i = 0; i < 100_000_000_000L; i++) {
        if (i % 1_000_000 == 0 && Thread.currentThread().isInterrupted()) {
            display("  computation interrupted at round " + i);
            return -1;
        }
        sum += i % 7;
    }
    return sum;
}

// display with the time elapsed since [start]
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async13.java
  318 ms :   block 1/10 received
  638 ms :   block 2/10 received
  939 ms :   block 3/10 received
 1212 ms :   download interrupted
 1214 ms : 1 - TimeoutException, state: CANCELLED
  301 ms : 2 - get() throws CancellationException, state: CANCELLED
  303 ms :   computation interrupted at round 131000000
  302 ms :   block 1/10 received
  603 ms :   block 2/10 received
  701 ms : 3 - the program requests the cancellation
  702 ms :   download interrupted
  702 ms : 3 - isCancelled=true
  309 ms :   block 1/3 received
  404 ms : 4 - CompletableFuture cancelled: isCancelled=true, but the operation continues
  609 ms :   block 2/3 received
  910 ms :   block 3/3 received

Comments:

• Lines 47–58: The simulated download receives a block every 300 ms. It is executed by a virtual thread and uses a blocking wait, [Thread.sleep], which terminates as soon as the thread is interrupted: [InterruptedException] exits the loop, and the download stops by signaling this. It requires no special parameters to be cancelable: the waits for [java.util.concurrent] react in the same way to the interruption, as do, in a virtual thread, the blocking I/O operations on sockets (the socket is then closed, and the operation initiates a [SocketException]);
• Lines 6–14: [get(1200, TimeUnit.MILLISECONDS)] waits for the result for at most 1.2 seconds, then launches a [TimeoutException]. The wait is abandoned but not the operation: it is explicitly canceled with [cancel(true)], which interrupts the download’s virtual thread. The thread has received three blocks and stops at 1,200 ms (lines 2–6 of the results);
• lines 16–26: a cancelable calculation (not a wait). The [compute] method (lines 60–71) checks its thread’s interrupt flag every million cycles and stops if it is set. The program aborts it after 300 ms: [get()] immediately launches a [CancellationException] (line 7), and the calculation stops on its own shortly thereafter (line 8);
• lines 28–35: The program requests cancellation after 700 ms. It takes effect immediately: the wait for the third block is interrupted (lines 11–13);
• Lines 37–43: A pitfall to be aware of. The method *[cancel] of a *[CompletableFuture] terminates the latter in the state *[CANCELLED], but does not interrupt anything: a [CompletableFuture] does not know which thread, if any, is working for it (its [mayInterruptIfRunning] parameter is ignored). The download continues until completion (lines 16–17); only its result will be ignored. For a [CompletableFuture] to be able to stop the operation it represents, that operation must monitor the state of the [CompletableFuture] ([isCancelled()]), or the [Future]* or the task’s thread must be retained to interrupt it;
• Line 44: The executor’s [close()] method waits for all tasks to complete, including the one in step 4.

9.6.2. Multiple processes, multiple outcomes

The program [Async14.java] launches five processes lasting 1 to 5 seconds: process No. 3 throws an exception, and process No. 4 is canceled. It processes each task as soon as it finishes, regardless of how it ended, and always knows which task it is:

// five processings of 1 to 5 seconds executed by virtual threads:
// no. 3 throws an exception, no. 4 is cancelled
long start;

void main() throws InterruptedException {
    start = System.nanoTime();
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
         ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor()) {
        // the CompletionService will deliver each task as soon as it is finished
        CompletionService<ProcessingResult> service = new ExecutorCompletionService<>(executor);
        List<Future<ProcessingResult>> tasks = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            int number = i;
            tasks.add(service.submit(() -> sleep(number, number + 1)));
        }
        // we cancel processing no. 4 after 2.5 seconds
        scheduler.schedule(() -> tasks.get(4).cancel(true), 2500, TimeUnit.MILLISECONDS);
        display("the " + tasks.size() + " processings are started");

        // we use each processing as soon as it is finished, whatever the way it finished
        for (int i = 0; i < tasks.size(); i++) {
            Future<ProcessingResult> task = service.take();
            int number = tasks.indexOf(task);
            String outcome = switch (task.state()) {
                case SUCCESS -> String.format("finished: scheduled duration %d s, actual duration %.1f s",
                        task.resultNow().duration(), task.resultNow().actualDuration());
                case FAILED -> "encountered the following error: " + task.exceptionNow().getMessage();
                case CANCELLED -> "cancelled";
                case RUNNING -> "still running?";
            };
            display("processing " + number + " " + outcome);
        }
    }
}

// the processing: sleeping [duration] seconds, in one-second slices
ProcessingResult sleep(int number, int duration) throws InterruptedException {
    long processingStart = System.nanoTime();
    // exception for processing no. 3
    if (number == 3) {
        throw new IllegalStateException("test....");
    }
    for (int i = 1; i <= duration; i++) {
        Thread.sleep(1000);
    }
    // the result returned to the caller
    return new ProcessingResult(duration, (System.nanoTime() - processingStart) / 1e9);
}

// the result of a processing
record ProcessingResult(int duration, double actualDuration) {
}

// display with the time elapsed since the start, in seconds
void display(String message) {
    IO.println(String.format("%4.1f s : %s", (System.nanoTime() - start) / 1e9, message));
}

Results:

> java Async14.java
 0.0 s : the 5 processings are started
 0.0 s : processing 3 encountered the following error: test....
 1.0 s : processing 0 finished: scheduled duration 1 s, actual duration 1.0 s
 2.0 s : processing 1 finished: scheduled duration 2 s, actual duration 2.0 s
 2.5 s : processing 4 cancelled
 3.0 s : processing 2 finished: scheduled duration 3 s, actual duration 3.0 s

Comments:

• Lines 36–48: the process. No. 3 throws an exception; the others “sleep” for [duration] seconds in one-second increments, with an interruptible [Thread.sleep], and return a result—a [record]—that contains the programmed duration and the actual duration. The method declares [InterruptedException]: it is called by a [Callable], whose method [call] can throw checked exceptions;
• lines 7–8: two executors declared within the same [try]: the virtual thread executor, and a [ScheduledExecutorService], an executor capable of executing a task after a delay (we’ll revisit this in the section on time);
• lines 10–15: the five processes are submitted to the [CompletionService] already encountered in [Async10.java], and their [Future] instances are stored in a list;
• line 17: [ScheduledExecutorService] will cancel process No. 4 after 2.5 seconds, even though it still has 2.5 seconds left to run;
• lines 21–32: [take()] delivers each process as soon as it is completed, regardless of how it ended: a canceled [Future] is also delivered. We do not call [get()], which would throw an exception, but instead examine the state of [Future] using a [switch] expression that covers the four constants of [Future.State] (the compiler verifies that none are missing). The processing number is determined by the position of its [Future] in the list ([indexOf]);
• the results show that No. 3 fails immediately, that Nos. 0, 1, and 2 terminate after 1, 2, and 3 seconds, respectively, and that No. 4 is canceled at 2.5 seconds.

9.6.3. Limiting the Duration of a Wait

The program [Async15.java] provides tools for limiting the duration of a wait: the methods [orTimeout] and [completeOnTimeout] from [CompletableFuture] (Java 9), and the method [get] with a delay from [Future]. It also shows how to combine two stop causes.

// limiting the duration of a wait: orTimeout, completeOnTimeout, get(timeout)
long start = System.nanoTime();

void main() throws InterruptedException, ExecutionException {
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        // 1 - orTimeout: we do not wait more than one second
        CompletableFuture<String> slow = CompletableFuture.supplyAsync(
                () -> queryBlocking("slow-server", 3000), executor);
        try {
            display("1 - answer: " + slow.orTimeout(1, TimeUnit.SECONDS).join());
        } catch (CompletionException e) {
            display("1 - " + e.getCause().getClass().getSimpleName() + ", state: " + slow.state());
        }
        // the operation itself continues: we let it finish
        Thread.sleep(2200);

        // 2 - completeOnTimeout: beyond one second, we settle for a default answer
        start = System.nanoTime();
        String answer = query("slow-server", 3000)
                .completeOnTimeout("default answer", 1, TimeUnit.SECONDS)
                .join();
        display("2 - " + answer);

        // 3 - to really stop the operation, we interrupt the thread executing it
        start = System.nanoTime();
        Future<String> slow3 = executor.submit(() -> queryBlocking("slow-server", 3000));
        try {
            display("3 - answer: " + slow3.get(1, TimeUnit.SECONDS));
        } catch (TimeoutException e) {
            slow3.cancel(true);
            display("3 - one-second timeout exceeded, operation cancelled, state: " + slow3.state());
        }
    }

    // 4 - two causes of stopping: the timeout specific to each processing and a general stop
    start = System.nanoTime();
    CompletableFuture<Void> generalStop = new CompletableFuture<>();
    CompletableFuture<Void> all = CompletableFuture.allOf(
            processWithTimeout("fast", 500, 1000, generalStop),
            processWithTimeout("slow", 3000, 1000, generalStop),
            processWithTimeout("medium", 1500, 2000, generalStop));
    // the caller requests the general stop after 1,200 ms
    Thread.sleep(1200);
    display("4 - general stop requested");
    generalStop.complete(null);
    all.join();
}

// a processing limited to [timeout] milliseconds, which can also be stopped by the caller
CompletableFuture<Void> processWithTimeout(String name, int duration, int timeout, CompletableFuture<Void> stop) {
    CompletableFuture<String> answer = query(name, duration).orTimeout(timeout, TimeUnit.MILLISECONDS);
    // the general stop cancels the wait for the answer
    stop.thenRun(() -> answer.cancel(false));
    // we examine the way the wait ended
    return answer.handle((text, e) -> switch (e) {
        case null -> text;
        case TimeoutException _ -> name + " : timeout of " + timeout + " ms exceeded";
        case CancellationException _ -> name + " : stopped at the caller's request";
        default -> name + " : " + e;
    }).thenAccept(message -> display("4 - " + message));
}

// simulates a server that answers after [duration] milliseconds
CompletableFuture<String> query(String server, int duration) {
    return new CompletableFuture<String>()
            .completeOnTimeout(server + " answered in " + duration + " ms", duration, TimeUnit.MILLISECONDS);
}

// the same operation, written in a blocking way
String queryBlocking(String server, int duration) {
    try {
        Thread.sleep(duration);
        display("  " + server + " has finished its work");
        return server + " answered in " + duration + " ms";
    } catch (InterruptedException e) {
        display("  " + server + " was interrupted");
        throw new CancellationException("operation interrupted");
    }
}

// display with the time elapsed since [start]
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async15.java
 1028 ms : 1 - TimeoutException, state: FAILED
 3025 ms :   slow-server has finished its work
 1001 ms : 2 - default answer
 1003 ms :   slow-server was interrupted
 1004 ms : 3 - one-second timeout exceeded, operation cancelled, state: CANCELLED
  501 ms : 4 - fast answered in 500 ms
 1005 ms : 4 - slow : timeout of 1000 ms exceeded
 1206 ms : 4 - general stop requested
 1207 ms : 4 - medium : stopped at the caller's request

Comments:

• lines 7–8: the slow operation is a blocking method (lines 69–79), executed by a virtual thread using [supplyAsync] with an executor;
• Line 10: [orTimeout(timeout, unit)] terminates [CompletableFuture] with a [TimeoutException] if it is not completed within the time limit. This is the direct write of the “run” with [anyOf] from the [Async09.java] program. [orTimeout] does not create a new [CompletableFuture]: it modifies the one to which it applies (and returns it); which is why the state of [slow] is [FAILED] (line 2 of the results);
• Lines 14–15: [orTimeout] aborts the wait, not the operation: the virtual thread continues to run and completes its work at 3 seconds (line 3), to no avail. This is the same issue seen earlier with [cancel]: a [CompletableFuture] cannot interrupt the operation it represents;
• lines 17–22: [completeOnTimeout(value, timeout, unit)] terminates [CompletableFuture] with a default value if it is not completed within the time limit: after one second, a “default answer” is returned (line 4);
• lines 24–32: To truly stop the operation, it is submitted to the executor; we wait for its [Future] for at most one second, then cancel it with [cancel(true)], which interrupts its thread: the operation stops after 1 second (lines 5–6). This is the preferred solution when the operation is blocking and interruptible;
• lines 49–61: a process may be stopped for several reasons; here, its own timeout and a general stop request from the caller:
  • line 51: the process’s own timeout is set with [orTimeout];
  • line 53: the general termination is itself a [CompletableFuture<Void>], which the caller will terminate; a continuation is attached to it that cancels the wait for the response;
  • lines 55–60: Regardless of the cause, the wait ends in failure; [handle] checks for an exception using a [switch] on the types (Java 21): [case null] for success, [case TimeoutException _] for timeout, [case CancellationException _] for general termination. The character [_] (Java 22) is an anonymous variable: the type is tested without needing to name the object;
• lines 36–46: three processes and a general shutdown requested at 1,200 ms upon completion of [CompletableFuture] and [generalStop]. Process [fast] completes normally; process [slow] exceeds its one-second time limit; process [medium], which was allotted two seconds, is terminated by the general shutdown (lines 7–10).

9.7. Reporting Progress

A long-running operation often needs to report its progress (a progress bar, a percentage). The simplest approach is to pass it a callback function that receives progress reports: a [Consumer<T>], the functional interface from Chapter 7 whose sole method, [accept(T)], takes a value. The operation calls [accept] at each step; it is unaware of how the progress reports will be displayed.

In a graphical user interface, the reports—generated by the thread executing the operation—cannot directly update the graphical components: [Consumer] must delegate the update to the UI thread ([SwingUtilities.invokeLater(() -> bar.setValue(percentage))] with Swing, [Platform.runLater] with JavaFX). In a console program, it can display the output directly. The program [Async16.java] simulates a file copy that the user can cancel by typing [Enter]:

// reporting the progress of an operation with a Consumer; the user can cancel it
void main() throws InterruptedException {
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        IO.println("Copy in progress, press [Enter] to cancel it");
        // the copy is executed by a virtual thread; the display method displays its reports
        Future<Integer> copy = executor.submit(() -> copy(8, this::display));
        // another virtual thread waits for the user to press [Enter] to cancel the copy
        Thread.ofVirtual().start(() -> {
            // readln returns null if the standard input is closed: we cancel nothing
            if (IO.readln() != null) {
                copy.cancel(true);
            }
        });
        try {
            IO.println("Copy finished: " + copy.get() + " files copied");
        } catch (CancellationException e) {
            IO.println("Copy cancelled by the user");
        } catch (ExecutionException e) {
            IO.println("Copy failed: " + e.getCause());
        }
    }
}

// the operation does not know how its progress is displayed: it only reports it
int copy(int nbFiles, Consumer<Progress> progress) throws InterruptedException {
    for (int i = 1; i <= nbFiles; i++) {
        // the copy of a file
        Thread.sleep(300);
        // the progress report (the caller may not want it: progress is then null)
        if (progress != null) {
            progress.accept(new Progress(100 * i / nbFiles, "file" + i + ".dat"));
        }
    }
    return nbFiles;
}

// a progress report
record Progress(int percentage, String file) {
}

// display of the progress reports in the console: a bar of 20 characters
void display(Progress progress) {
    int nbHashes = progress.percentage() / 5;
    String bar = "#".repeat(nbHashes) + ".".repeat(20 - nbHashes);
    IO.println(String.format("[%s] %3d %% %s", bar, progress.percentage(), progress.file()));
}

Results without cancellation:

> java Async16.java
Copy in progress, press [Enter] to cancel it
[##..................]  12 % file1.dat
[#####...............]  25 % file2.dat
[#######.............]  37 % file3.dat
[##########..........]  50 % file4.dat
[############........]  62 % file5.dat
[###############.....]  75 % file6.dat
[#################...]  87 % file7.dat
[####################] 100 % file8.dat
Copy finished: 8 files copied

Results when the user enters [Enter] during the copy operation, after about one and a half seconds:

> java Async16.java
Copy in progress, press [Enter] to cancel it
[##..................]  12 % file1.dat
[#####...............]  25 % file2.dat
[#######.............]  37 % file3.dat
[##########..........]  50 % file4.dat
[############........]  62 % file5.dat
Copy cancelled by the user

Comments:

• Lines 24–35: The operation receives a [Consumer<Progress>]. After each file, it reports its progress using a [Progress] object (a [record] that includes the percentage and the file name). A caller who does not want reports passes [null], hence the check on line 30;
• lines 41–46: displaying a report: a 20-character bar constructed using [String.repeat] (Java 11). The method reference [this::display] (line 6) is the [Consumer<Progress>] passed to the operation;
• line 6: the copy is performed by a virtual thread;
• lines 8–13: another virtual thread waits for the user to type [Enter] ([IO.readln()] is a blocking read, which is why this thread: it is this thread that is blocked, not the program), then cancels the copy with [cancel(true)], which interrupts the [Thread.sleep] on line 28. The program does not wait for it: it is a daemon thread, abandoned at the end of the program if the user has not entered anything;
• lines 14–20: the program waits for the copy to finish. A cancellation results in a [CancellationException], while a copy failure results in a [ExecutionException].

9.8. Asynchronous Input/Output

9.8.1. Files

I/O is the very reason for asynchronous processing. Java provides two families of classes for files:

• the “classic,” blocking classes: [Files] ([readAllLines], [readString], [lines], [writeString], [newBufferedReader], [newBufferedWriter]…), [BufferedReader], [BufferedWriter]. These are now executed on a virtual thread when you do not want to block an important thread;
• the [AsynchronousFileChannel] class (Java 7), whose operations return a [Future] or call a [CompletionHandler]. It is rarely used: on Linux, it actually performs blocking reads on a thread pool internal to JVM, and its interface is cumbersome.

The program [Async17.java] displays both families by reading the tax bracket file from the main program, [TaxData.txt], located in the program’s directory:

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

Each line contains the bracket threshold, the coefficient [coeffR] and the coefficient [coeffN], separated by the character [:] and written with a decimal point (the tax scale is the 2004 scale for 2003 income, retained as-is for this exercise). The program runs in the [09-async] folder and searches for its files in the current folder.

// file input / output: AsynchronousFileChannel, then blocking I/O on a virtual thread
final Path data = Path.of("TaxData.txt");
final Path report = Path.of("report-17.txt");

void main() throws InterruptedException {
    // 1 - AsynchronousFileChannel (Java 7): an asynchronous read, returned as a CompletableFuture
    String content = readAsynchronously(data).join();
    IO.println("1 - AsynchronousFileChannel: " + content.lines().count() + " lines");

    // 2 to 7 - ordinary blocking I/O, executed by a virtual thread
    Thread.ofVirtual().start(() -> {
        try {
            processFiles();
        } catch (IOException e) {
            IO.println("error: " + e);
        }
    }).join();
}

// reads a whole (small) file with an AsynchronousFileChannel
CompletableFuture<String> readAsynchronously(Path file) {
    CompletableFuture<String> result = new CompletableFuture<>();
    try {
        AsynchronousFileChannel channel = AsynchronousFileChannel.open(file, StandardOpenOption.READ);
        ByteBuffer buffer = ByteBuffer.allocate((int) channel.size());
        // the read is started; the callback object will complete the CompletableFuture
        channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer nbBytes, ByteBuffer read) {
                result.complete(new String(read.array(), 0, nbBytes, StandardCharsets.UTF_8));
            }

            @Override
            public void failed(Throwable e, ByteBuffer read) {
                result.completeExceptionally(e);
            }
        });
        // the channel will be closed when the read is finished, whatever its outcome
        result.whenComplete((text, e) -> close(channel));
    } catch (IOException e) {
        result.completeExceptionally(e);
    }
    return result;
}

void processFiles() throws IOException {
    // 2 - reading the whole file in a single operation
    List<String> lines = Files.readAllLines(data);
    IO.println("2 - Files.readAllLines: " + lines.size() + " lines");

    // 3 - reading line by line with a BufferedReader
    List<TaxBracket> brackets = new ArrayList<>();
    try (BufferedReader reader = Files.newBufferedReader(data)) {
        String line;
        // readLine returns null at the end of the file
        while ((line = reader.readLine()) != null) {
            brackets.add(TaxBracket.parse(line));
        }
    }
    IO.println("3 - BufferedReader.readLine: " + brackets.size() + " brackets, 2nd: " + brackets.get(1));

    // 4 - Files.lines returns a Stream of lines, read on demand
    try (Stream<String> stream = Files.lines(data)) {
        IO.println("4 - Files.lines: " + stream.count() + " lines");
    }

    // 5 - writing with a BufferedWriter
    try (BufferedWriter writer = Files.newBufferedWriter(report)) {
        writer.write("Brackets of the tax scale (exercise: 2004 tax scale on 2003 income)");
        writer.newLine();
        for (int i = 0; i < brackets.size(); i++) {
            TaxBracket bracket = brackets.get(i);
            String limit = bracket.limit().signum() == 0 ? "beyond"
                    : String.format("up to %,d €", bracket.limit().intValue());
            writer.write(String.format("bracket %d : %-18s rate %.1f %%",
                    i + 1, limit, bracket.coeffR().multiply(BigDecimal.valueOf(100))));
            writer.newLine();
        }
    }

    // 6 - reading the whole file as a string
    String text = Files.readString(report);
    IO.print("5 and 6 - content of [" + report + "] :\n" + text);

    // 7 - appending at the end of the file then deleting it
    Files.writeString(report, "end of the report\n", StandardOpenOption.APPEND);
    IO.println("7 - last line: " + Files.readAllLines(report).getLast());
    Files.delete(report);
}

// closes a channel without checked exception
void close(Channel channel) {
    try {
        channel.close();
    } catch (IOException e) {
        IO.println("error while closing: " + e);
    }
}

// a tax bracket; the file uses the decimal point, which BigDecimal understands
record TaxBracket(BigDecimal limit, BigDecimal coeffR, BigDecimal coeffN) {
    static TaxBracket parse(String line) {
        String[] fields = line.split(":");
        return new TaxBracket(new BigDecimal(fields[0]), new BigDecimal(fields[1]),
                new BigDecimal(fields[2]));
    }
}

Results:

> java Async17.java
1 - AsynchronousFileChannel: 7 lines
2 - Files.readAllLines: 7 lines
3 - BufferedReader.readLine: 7 brackets, 2nd: TaxBracket[limit=8382, coeffR=0.068, coeffN=291.09]
4 - Files.lines: 7 lines
5 and 6 - content of [report-17.txt] :
Brackets of the tax scale (exercise: 2004 tax scale on 2003 income)
bracket 1 : up to 4,962 €      rate 0.0 %
bracket 2 : up to 8,382 €      rate 6.8 %
bracket 3 : up to 14,753 €     rate 19.1 %
bracket 4 : up to 23,888 €     rate 28.3 %
bracket 5 : up to 38,868 €     rate 37.4 %
bracket 6 : up to 47,932 €     rate 42.6 %
bracket 7 : beyond             rate 48.1 %
7 - last line: end of the report

Comments:

• Lines 20–44: Reading with a [AsynchronousFileChannel], presented in the form of a [CompletableFuture<String>]. This is the general technique for converting a API with callback methods into a [CompletableFuture]: we create a “work-in-progress” [CompletableFuture] (line 22), return it immediately (line 43), and the callback methods complete it, returning either the result ([complete], line 30) or the exception ([completeExceptionally], line 35). The callback object is an anonymous class that implements [CompletionHandler<Integer, ByteBuffer>]: the result of a read operation is the number of bytes read, and the object passed here is the buffer itself. A single read operation is sufficient for a small file; a large file would require successive read operations, which the callback methods would need to chain together;
• line 39: [whenComplete] closes the channel when the read operation is complete, whether it was successful or not;
• Lines 11–17: The other operations are ordinary blocking I/O operations, grouped within the [processFiles] method and executed by a virtual thread. The lambda of a [Runnable] cannot throw a checked exception: the [IOException] is intercepted within the lambda;
• line 48: [Files.readAllLines] reads the entire file and returns a list of its lines (using UTF-8 by default, like all methods of [Files]);
• lines 52–59: line-by-line reading using a [BufferedReader]; [readLine()] returns [null] at the end of the file. Each line is converted to [TaxBracket] by the record’s [parse] method (lines 100–107); the numbers are [BigDecimal], whose constructor expects a decimal point regardless of the machine’s language. The representation of a record produced by its [toString] method appears on line 4 of the results;
• lines 63–65: [Files.lines] returns a [Stream<String>] consisting of lines read on demand: unlike [readAllLines], it does not load the entire file into memory. This is the preferred method for large files. The stream keeps the file open: it must be closed, hence the [try] with a resource;
• Lines 68–79: Writing with a [BufferedWriter], which is closed (and thus flushed to disk) upon exit from the [try]. The [%,d] format inserts the thousands separator for the machine’s language (a thin non-breaking space in French);
• Lines 82–88: [Files.readString] (Java 11) re-reads the written file; [Files.writeString], using the [StandardOpenOption.APPEND] option, adds a line to it; [getLast()] (Java 21) returns the last element of the list, then the file is deleted.

Recall the limitation mentioned above: file reads and writes pin the virtual thread to its carrier thread during the operation, and JVM compensates by adding a carrier thread. For local files, this is rarely noticeable; for network operations, virtual threads actually release their carriers during waits.

9.8.2. Network: HttpClient

The Java client HTTP, [java.net.http.HttpClient] (Java 11), offers both approaches: [send], which is blocking and returns the response, and [sendAsync], which immediately returns a [CompletableFuture] containing the response. Chapter 12 will examine this in detail; the program [Async18.java] simply compares the two methods by querying a small local server, HTTP, which takes one second to return the “temperature” of a city:

// HttpClient: asynchronous (sendAsync) or blocking (send) HTTP requests to a small local server
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> cities = List.of("Angers", "Brest", "Colmar", "Dijon", "Évry", "Foix");
long start;

void main() throws IOException, InterruptedException, ExecutionException {
    // an HTTP server that takes one second to give the "temperature" of a city
    HttpServer server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), 0);
    server.createContext("/temperature", exchange -> {
        // the request has the form /temperature?city=xxx
        String parameter = exchange.getRequestURI().getRawQuery().substring("city=".length());
        String city = URLDecoder.decode(parameter, StandardCharsets.UTF_8);
        sleep(1000);
        byte[] response = (city + " : " + (10 + city.length()) + " °C").getBytes(StandardCharsets.UTF_8);
        exchange.sendResponseHeaders(200, response.length);
        try (OutputStream stream = exchange.getResponseBody()) {
            stream.write(response);
        }
    });
    // each received request is handled by a virtual thread
    server.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
    server.start();

    try (HttpClient client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build()) {
        // a first request, so that the JVM loads the client classes before the measurements
        client.send(request("Paris"), HttpResponse.BodyHandlers.discarding());

        // 1 - sendAsync immediately returns a CompletableFuture<HttpResponse<String>>
        start = System.nanoTime();
        List<CompletableFuture<String>> responses = cities.stream()
                .map(city -> client.sendAsync(request(city), HttpResponse.BodyHandlers.ofString())
                        .thenApply(HttpResponse::body))
                .toList();
        CompletableFuture.allOf(responses.toArray(CompletableFuture[]::new)).join();
        display("1 - sendAsync: " + responses.stream().map(CompletableFuture::resultNow).toList());

        // 2 - send is blocking: we call it from virtual threads
        start = System.nanoTime();
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> responses2 = new ArrayList<>();
            for (String city : cities) {
                responses2.add(executor.submit(
                        () -> client.send(request(city), HttpResponse.BodyHandlers.ofString()).body()));
            }
            List<String> texts = new ArrayList<>();
            for (Future<String> response : responses2) {
                texts.add(response.get());
            }
            display("2 - send on virtual threads: " + texts);
        }
    } finally {
        server.stop(0);
    }
}

// the GET request that asks for the temperature of a city
HttpRequest request(String city) {
    String parameter = URLEncoder.encode(city, StandardCharsets.UTF_8);
    URI address = URI.create("http://localhost:" + port + "/temperature?city=" + parameter);
    return HttpRequest.newBuilder(address).build();
}

// Thread.sleep without checked exception: an interruption becomes a CancellationException
void sleep(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("wait interrupted");
    }
}

// display with the time elapsed since [start]
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> 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 on virtual threads: [Angers : 16 °C, Brest : 15 °C, Colmar : 16 °C, Dijon : 15 °C, Évry : 14 °C, Foix : 14 °C]

Comments:

• lines 12–27: the server. The JVM file contains a small HTTP server, the [HttpServer] class from the [jdk.httpserver] module, imported on line 2 (a compact file automatically imports only the [java.base] module; the client classes, from the [java.net.http] module, are imported on lines 3–5). [createContext] associates a handler with the [/temperature] path—a lambda that receives the HTTP exchange: it decodes the [city] parameter from the requested address, waits one second, and then sends the response. Line 26 has each request processed by a virtual thread: a server configured this way can serve thousands of waiting clients;
• Lines 29–31: The client is created by a constructor ([newBuilder]) that sets the protocol version here. Since Java 21, it implements [AutoCloseable] and declares itself in a [try] with a resource. An initial request is used to “warm up” the JVM: the first time a client is used, it loads numerous classes, which would skew the subsequent measurement by a few hundred milliseconds;
• lines 33–40: [sendAsync(request, handler)] sends the request and immediately returns a [CompletableFuture<HttpResponse<String>>]; the body handler [BodyHandlers.ofString()] indicates that the response body must be read as a string. [thenApply(HttpResponse::body)] retains only the body. The six requests are sent simultaneously, and their responses arrive together after one second (line 2 of the results). While waiting, no thread waits for the response: the client monitors its connections using a “selector” thread;
• Lines 42–55: [send] is blocking; it is being called from six virtual threads. The result is the same: one second for all six requests (line 3), and no system threads are blocked;
• lines 61–66: the request GET. [URLEncoder.encode] encodes characters that cannot appear as-is in an address (the [É] from Évry), and the server decodes them using [URLDecoder.decode] (line 17).

9.9. Asynchronous Value Streams

An asynchronous operation returns a result after a certain amount of time. A stream ([Stream]) or an iterator returns a sequence of values, produced on demand. Sometimes it is necessary to combine the two: a sequence of values where each one requires a wait before being produced. This is the case for lines in a large file, the rows in the result of a query SQL, messages arriving over a network connection, sensor readings… Other languages have a dedicated type for this; Java offers two models:

• the “pull” model: the consumer requests the next value and waits for it to be available. With virtual threads, this is simply a [Iterator] or a [Stream] where the production of each value blocks, read by a virtual thread;
• the “push” model: the producer sends values to its subscribers as they are produced, and the subscribers control the throughput. This is the reactive stream model, standardized by the [java.util.concurrent.Flow] interfaces (Java 9).

9.9.1. The pull model: a Stream read by a virtual thread

The program [Async19.java] produces measurements—one every 200 ms—in the form of a [Stream<Measurement>]:

// a sequence of values produced with waits: a "blocking" Stream read by a virtual thread
long start;

void main() throws InterruptedException {
    // all the work is done by a virtual thread: its waits block no carrier thread
    Thread.ofVirtual().start(this::consume).join();
}

void consume() {
    start = System.nanoTime();
    // 1 - each value is processed as soon as it is produced
    try (Stream<Measurement> measurements = readMeasurements("sensor-1", 4)) {
        measurements.forEach(measurement -> display("1 - received " + measurement));
    }

    // 2 - we can leave the loop before the end: the producer does not produce the following values
    start = System.nanoTime();
    try (Stream<Measurement> measurements = readMeasurements("sensor-2", 100)) {
        Iterator<Measurement> iterator = measurements.iterator();
        while (iterator.hasNext()) {
            Measurement measurement = iterator.next();
            display("2 - received " + measurement);
            if (measurement.value() >= 22) {
                break;
            }
        }
    }

    // 3 - cancellation: another thread interrupts the thread reading the stream
    start = System.nanoTime();
    Thread reader = Thread.ofVirtual().start(() -> {
        try (Stream<Measurement> measurements = readMeasurements("sensor-3", 100)) {
            measurements.forEach(measurement -> display("3 - received " + measurement));
        } catch (CancellationException e) {
            display("3 - reading cancelled");
        }
    });
    waitForEnd(reader, 700);

    // 4 - all the operations of the Stream API apply to the stream
    start = System.nanoTime();
    String high;
    try (Stream<Measurement> measurements = readMeasurements("sensor-4", 6)) {
        high = measurements
                .filter(measurement -> measurement.value() > 21)
                .map(measurement -> NumberFormat.getInstance().format(measurement.value()))
                .collect(Collectors.joining(" ; ", "[", "]"));
    }
    display("4 - measurements above 21: " + high);
}

// a producer: one measurement every 200 ms, produced only when the consumer asks for it
Stream<Measurement> readMeasurements(String sensor, int count) {
    return IntStream.rangeClosed(1, count)
            .mapToObj(i -> {
                // waiting for the next measurement
                sleep(200);
                return new Measurement(sensor, i, 19.5 + i * 0.75);
            })
            // executed by close(): at the normal end, after a break or on cancellation
            .onClose(() -> IO.println("          " + sensor + " : end of production"));
}

// waits [timeout] ms for the end of a thread, then interrupts it and waits for its end
void waitForEnd(Thread thread, long timeout) {
    try {
        thread.join(timeout);
        thread.interrupt();
        thread.join();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

// Thread.sleep without checked exception: an interruption becomes a CancellationException
void sleep(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("wait interrupted");
    }
}

// display with the time elapsed since [start]
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

// a measurement
record Measurement(String sensor, int number, double value) {
    @Override
    public String toString() {
        return sensor + " no. " + number + " : " + NumberFormat.getInstance().format(value) + " °C";
    }
}

Results:

> java Async19.java
  249 ms : 1 - received sensor-1 no. 1 : 20.25 °C
  455 ms : 1 - received sensor-1 no. 2 : 21 °C
  656 ms : 1 - received sensor-1 no. 3 : 21.75 °C
  857 ms : 1 - received sensor-1 no. 4 : 22.5 °C
          sensor-1 : end of production
  200 ms : 2 - received sensor-2 no. 1 : 20.25 °C
  401 ms : 2 - received sensor-2 no. 2 : 21 °C
  603 ms : 2 - received sensor-2 no. 3 : 21.75 °C
  804 ms : 2 - received sensor-2 no. 4 : 22.5 °C
          sensor-2 : end of production
  201 ms : 3 - received sensor-3 no. 1 : 20.25 °C
  402 ms : 3 - received sensor-3 no. 2 : 21 °C
  602 ms : 3 - received sensor-3 no. 3 : 21.75 °C
          sensor-3 : end of production
  702 ms : 3 - reading cancelled
          sensor-4 : end of production
 1206 ms : 4 - measurements above 21: [21.75 ; 22.5 ; 23.25 ; 24]

Comments:

• Lines 52–62: the producer. It returns a [Stream<Measurement>] constructed from the integers 1 through [count]; the [mapToObj] function waits 200 ms and then returns a measurement. Since a stream is lazy, this function is called only when the consumer requests the next value, and each measurement is processed before the next one is produced. [onClose] records an action executed when the stream is closed, regardless of the reason: this is the producer’s [finally];
• line 6: all the work is done by a virtual thread: the 200 ms wait for each measurement does not block any system threads;
• Lines 11–14: Each measurement is processed as soon as it is produced: one every 200 ms (lines 2–5 of the results). The stream is declared in a [try] with a resource: closing it triggers the [onClose] action (line 6);
• lines 16–27: To exit the loop before it ends, the stream is traversed using its iterator. After the [break], the producer does not produce the remaining 96 measurements, and the stream is closed upon exiting the [try] (line 11). The methods [takeWhile] (Java 9) or [limit] also allow you to stop a stream;
• lines 29–38: cancellation. The stream is read by another virtual thread, which the [waitForEnd] method (lines 64–73) interrupts after 700 ms. The wait for the fourth measurement is interrupted: the [sleep] method invokes a [CancellationException], which traverses the stream; the [try] with a resource closes the stream (line 15 of the results), then the [catch] reports the cancellation (line 16);
• lines 40–49: everything the API Stream can do is applied to the stream: filtering, transformation, collection. The result is only available at the end of the stream, after 6 × 200 ms;
• lines 90–96: the [toString] method of the [Measurement] record is redefined for a more compact display; [NumberFormat.getInstance()] formats the real number according to the machine’s language, without unnecessary decimal places ([21], [21.75]).

The same pattern applies to a [BufferedReader] (its lines are read on demand, and its method [lines()] returns a [Stream<String>]), to a [ResultSet], JDBC, or to a socket’s incoming stream: on a virtual thread, a blocking stream is an asynchronous stream.

9.9.2. The push model: Flow and SubmissionPublisher

The [java.util.concurrent.Flow] interface (Java 9) groups together four internal interfaces that standardize reactive streams:

• [Flow.Publisher<T>], the publisher, produces values; its method [subscribe] adds a subscriber to it;
• [Flow.Subscriber<T>], the subscriber, receives values via its method [onNext], the end of the stream via [onComplete], an error via [onError], and at the beginning, the object [Flow.Subscription], which links it to the publisher via [onSubscribe];
• [Flow.Subscription], the subscription: [request(n)] requests additional values from [n], and [cancel()] terminates the subscription. The publisher never sends more values than the subscriber has requested: this is “backpressure,” which prevents a fast publisher from overwhelming a slow subscriber;
• [Flow.Processor<T, R>] is both a subscriber and a publisher (a transformation node).

Java provides an implementation of the publisher, [SubmissionPublisher<T>], which delivers values to each subscriber via an executor and a buffer specific to that subscriber. Reactive libraries (Project Reactor, RxJava) implement these same interfaces and add dozens of operators to them. The program [Async20.java] publishes six metrics to two subscribers: a fast display that unsubscribes when the temperature exceeds 22 °C, and a slow archive:

// the "push" model: java.util.concurrent.Flow and SubmissionPublisher (Java 9)
final long start = System.nanoTime();

void main() throws InterruptedException {
    Subscriber screen = new Subscriber("screen", 0, 22);
    Subscriber archive = new Subscriber("archive", 500, Double.MAX_VALUE);
    // the publisher gives the delivery of the values to an executor and keeps a buffer of 2 values per subscriber
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var publisher = new SubmissionPublisher<Measurement>(executor, 2);
        publisher.subscribe(screen);
        publisher.subscribe(archive);
        for (int i = 1; i <= 6; i++) {
            sleep(200);
            // submit waits if there is no room left in a subscriber's buffer; it returns the estimated lag
            int lag = publisher.submit(new Measurement("sensor-1", i, 19.5 + i * 0.75));
            display("publisher: measurement no. " + i + " published, estimated lag: " + lag);
        }
        // close() signals the end of the stream to the subscribers: their onComplete method will be called
        display("publisher: end of publication");
        publisher.close();
        // we wait for the archive to have processed everything
        archive.end.await();
    }
}

// a subscriber that processes each measurement in [duration] ms and unsubscribes beyond [threshold]
class Subscriber implements Flow.Subscriber<Measurement> {
    private final String name;
    private final long duration;
    private final double threshold;
    private Flow.Subscription subscription;
    final CountDownLatch end = new CountDownLatch(1);

    Subscriber(String name, long duration, double threshold) {
        this.name = name;
        this.duration = duration;
        this.threshold = threshold;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        // the subscriber requests the values one by one: it is the one that controls the rate
        subscription.request(1);
    }

    @Override
    public void onNext(Measurement measurement) {
        display("  " + name + " : received " + measurement);
        sleep(duration);
        if (measurement.value() >= threshold) {
            display("  " + name + " : threshold reached, unsubscribing");
            subscription.cancel();
        } else {
            subscription.request(1);
        }
    }

    @Override
    public void onError(Throwable e) {
        display("  " + name + " : error " + e);
        end.countDown();
    }

    @Override
    public void onComplete() {
        display("  " + name + " : end of the stream");
        end.countDown();
    }
}

// Thread.sleep without checked exception: an interruption becomes a CancellationException
void sleep(long ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new CancellationException("wait interrupted");
    }
}

// display with the time elapsed since the start
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

// a measurement
record Measurement(String sensor, int number, double value) {
    @Override
    public String toString() {
        return sensor + " no. " + number + " : " + NumberFormat.getInstance().format(value) + " °C";
    }
}

Results:

> java Async20.java
  221 ms : publisher: measurement no. 1 published, estimated lag: 1
  269 ms :   archive : received sensor-1 no. 1 : 20.25 °C
  274 ms :   screen : received sensor-1 no. 1 : 20.25 °C
  466 ms : publisher: measurement no. 2 published, estimated lag: 2
  466 ms :   screen : received sensor-1 no. 2 : 21 °C
  667 ms :   screen : received sensor-1 no. 3 : 21.75 °C
  667 ms : publisher: measurement no. 3 published, estimated lag: 3
  770 ms :   archive : received sensor-1 no. 2 : 21 °C
  867 ms : publisher: measurement no. 4 published, estimated lag: 3
  868 ms :   screen : received sensor-1 no. 4 : 22.5 °C
  868 ms :   screen : threshold reached, unsubscribing
 1271 ms :   archive : received sensor-1 no. 3 : 21.75 °C
 1271 ms : publisher: measurement no. 5 published, estimated lag: 3
 1771 ms :   archive : received sensor-1 no. 4 : 22.5 °C
 1772 ms : publisher: measurement no. 6 published, estimated lag: 3
 1772 ms : publisher: end of publication
 2274 ms :   archive : received sensor-1 no. 5 : 23.25 °C
 2775 ms :   archive : received sensor-1 no. 6 : 24 °C
 3276 ms :   archive : end of the stream

Comments:

• lines 26–70: the subscriber, which processes each measurement in [duration] milliseconds and unsubscribes when the measurement reaches [threshold]:
  • lines 40–45: [onSubscribe] receives the subscription and requests an initial value;
  • lines 47–57: [onNext] processes a value and then either requests another one ([request(1)]) or terminates the subscription ([cancel()]). By requesting values one at a time, the subscriber controls the data rate themselves;
  • lines 59–69: [onError] and [onComplete] signal the end of the stream, either due to an error or normally. [CountDownLatch] allows the main program to wait for this end;
• line 9: The publisher delivers the values to each subscriber via a task of the specified executor—in this case, virtual threads (without an executor, it would use the shared pool, where waiting for the archive would block one of its few threads)—and maintains a buffer of up to 2 values for each subscriber;
• lines 12–17: the publisher publishes a measurement every 200 ms using [submit], which returns an estimate of the delay: the largest number of values published but not yet consumed by a subscriber. When a subscriber’s buffer is full, [submit] waits for a slot to become available: measurement #5, ready at around 1,070 ms, is not published until 1,271 ms (line 14 of the results), when the archive has processed measurement #3. This is backpressure: the publisher is slowed down to the pace of the slowest subscriber. ([SubmissionPublisher] also offers [offer], which allows a value to be discarded instead of waiting);
• the results show that the screen, which is fast, processes each measurement as soon as it is published and unsubscribes at measurement No. 4 (22.5 °C, lines 11–12), while the archive, which takes 500 ms per measurement, falls behind and receives all six measurements;
• lines 18–22: [close()] signals the end of the stream: each subscriber still subscribed will receive [onComplete] after processing the values in its buffer (line 20). The program waits for the end of the archive with the message [CountDownLatch].

The push model is suitable for sources that produce data at their own pace (sensors, events, messages) and for reactive processing chains; the pull model, which is simpler to write and read, is suitable for reading from a source on demand. With virtual threads, the pull model is usually sufficient.

9.10. Cache, switch back to the correct thread, and finish the operation yourself

9.10.1. A cache for asynchronous operations

A [CompletableFuture] is an object: each call to a method that returns one allocates at least one object on the heap, which the garbage collector must reclaim. This is negligible for I/O that lasts milliseconds, and current garbage collectors in JVM handle short-lived objects very efficiently: Java does not have a “lightweight” variant of [CompletableFuture] (other languages do have one, for asynchronous methods called millions of times that most often already have their result).

For an asynchronous method that often already has the result, best practice is to cache the [CompletableFuture] objects themselves: a value that is already known is returned by a [CompletableFuture] that has already completed, without a new allocation, and multiple concurrent requests for the same value share the same ongoing operation. The [Async21.java] program applies this to an exchange rate service:

// a cache of CompletableFutures: an asynchronous operation that, most of the time, already has the result
void main() {
    var service = new RateService();
    // 1 - successive readings: only the first reading of a currency queries the server
    for (String code : List.of("EUR", "USD", "EUR", "EUR", "USD", "GBP", "EUR")) {
        long readingStart = System.nanoTime();
        BigDecimal rate = service.readRate(code).join();
        IO.println(String.format("%s : %6s read in %3d ms", code, NumberFormat.getInstance().format(rate),
                (System.nanoTime() - readingStart) / 1_000_000));
    }
    IO.println("1 - server accesses: " + service.nbServerAccesses() + ", readings: " + service.nbReadings());

    // 2 - three simultaneous readings of the same currency: they share the same operation
    List<CompletableFuture<BigDecimal>> readings = List.of(
            service.readRate("CHF"), service.readRate("CHF"), service.readRate("CHF"));
    IO.println("2 - same object: " + (readings.get(0) == readings.get(2))
            + ", rate: " + NumberFormat.getInstance().format(readings.get(0).join())
            + ", server accesses: " + service.nbServerAccesses() + ", readings: " + service.nbReadings());

    // 3 - a failed operation is not kept in the cache
    try {
        service.readRate("XYZ").join();
    } catch (CompletionException e) {
        IO.println("3 - " + e.getCause().getMessage() + ", cached currencies: " + service.cachedCurrencies());
    }
}

class RateService {
    // the cache contains the CompletableFutures themselves, completed or in progress
    private final Map<String, CompletableFuture<BigDecimal>> cache = new ConcurrentHashMap<>();
    private final AtomicInteger nbServerAccesses = new AtomicInteger();
    private final AtomicInteger nbReadings = new AtomicInteger();

    CompletableFuture<BigDecimal> readRate(String code) {
        nbReadings.incrementAndGet();
        // if the currency is in the cache, we return the existing CompletableFuture: no object is created
        return cache.computeIfAbsent(code, this::readRateFromServer);
    }

    private CompletableFuture<BigDecimal> readRateFromServer(String code) {
        nbServerAccesses.incrementAndGet();
        // simulates querying a remote server (300 ms)
        CompletableFuture<BigDecimal> reading = new CompletableFuture<Void>()
                .completeOnTimeout(null, 300, TimeUnit.MILLISECONDS)
                .thenApply(nothing -> 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("unknown currency: " + code);
                });
        // in case of failure, we remove the operation from the cache so that a later reading is retried
        return reading.whenComplete((rate, e) -> {
            if (e != null) {
                cache.remove(code);
            }
        });
    }

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

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

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

Results:

> java Async21.java
EUR :      1 read in 363 ms
USD :   1.17 read in 300 ms
EUR :      1 read in   0 ms
EUR :      1 read in   0 ms
USD :   1.17 read in   0 ms
GBP :   0.87 read in 300 ms
EUR :      1 read in   0 ms
1 - server accesses: 3, readings: 7
2 - same object: true, rate: 0.94, server accesses: 4, readings: 10
3 - unknown currency: XYZ, cached currencies: [CHF, EUR, GBP, USD]

Comments:

• line 30: The cache associates each currency code with the [CompletableFuture] for its exchange rate, whether completed or still in progress. This is a [ConcurrentHashMap] (Chapter 8), because it can be used by multiple threads at the same time (the main thread, the timer thread that completes the reads);
• Lines 34–38: [computeIfAbsent(key, function)] returns the value associated with the key if it exists; otherwise, it calls the function, stores its result in the dictionary, and returns it. For a [ConcurrentHashMap], this operation is atomic: two threads that simultaneously request a currency that is not present trigger only a single access to the server;
• lines 40–58: the simulated server query (300 ms). The continuation [thenApply] contains a [switch] expression on a string, whose branch [default] throws an exception for an unknown currency;
• lines 52–57: in case of failure, the operation is removed from the cache so that a subsequent request for the same currency will query the server again instead of indefinitely returning the same failure. It is the [CompletableFuture] returned by [whenComplete] that is stored in the cache: it does not complete until after the action has been executed, which guarantees the caller that the cache is up to date when they receive the result;
• lines 5–11: For the caller, nothing changes: it waits for the result with [join()]. The results show the three server accesses (300 ms each; slightly longer for the first one, during which JVM loads the necessary classes) and the four immediate responses from the cache;
• lines 13–18: Three simultaneous requests for the Swiss franc exchange rate retrieve the same object and result in only one server access (line 10 of the results);
• lines 20–25: the unknown currency resulted in a failure and is no longer in the cache (line 11).

9.10.2. Return to the GUI thread

The program [Async06.java] demonstrated that a continuation is executed, depending on the case, either by the thread that completes the operation or by a thread of the selected executor. In a graphical application (Swing, JavaFX), this matters: a graphical component can only be manipulated by the UI thread. Java has no mechanism that automatically returns the rest of the code to this thread; it is up to the program to request this, either by passing the correct executor to the continuation or by using the tools of the graphics library:

• with Swing, *[SwingUtilities.invokeLater(action)] causes the action to be executed by the interface thread; the method reference [SwingUtilities::invokeLater] can serve as an executor for a continuation: [download.thenAcceptAsync(page -> textArea.setText(page), SwingUtilities::invokeLater)]******. The [SwingWorker] class also provides a method, [doInBackground], which is executed by a background thread, and methods [process] and [done]*, which are executed by the interface’s thread;
• along with JavaFX, [Platform.runLater(action)] serves the same purpose, and the classes [Task] and [Service] of [javafx.concurrent] encapsulate a background process along with its progress.

In all cases, the rule is the same: the UI thread must never wait ([join()], [get()], blocking I/O), otherwise the window freezes. The long-running task is executed by another thread (a virtual thread, for example), and only the final update of the components is handled by the interface thread.

9.10.3. Terminating a CompletableFuture Manually

Sometimes you may need to use a API that signals its events via listeners (Chapter 7), while the rest of the code is written using [CompletableFuture]. We saw the technique with [Async17.java]: create a “pending” [CompletableFuture] using its constructor, return it immediately, and terminate it yourself when needed:

• [complete(value)] completes it successfully;
• [completeExceptionally(exception)] fails to complete it;
• [cancel(false)] cancels it.

These three methods have no effect if [CompletableFuture] has already been completed ([complete] and [completeExceptionally] then produce [false]): the first one to complete it wins, which naturally resolves contention situations (such as a result and a timeout arriving at the same time). The program [Async22.java] waits for a thermometer—which notifies its listeners with each measurement—to reach a certain temperature:

// a CompletableFuture completed "by hand": turning a listener-based API into an asynchronous operation
final long start = System.nanoTime();

void main() {
    var thermometer = new Thermometer();
    thermometer.start();

    // 1 - we wait for the first measure greater than or equal to 21 °C
    double measure = waitForTemperature(thermometer, 21).join();
    display("1 - temperature reached: " + format(measure) + " °C");

    // 2 - the same wait, limited to one second
    try {
        measure = waitForTemperature(thermometer, 30).orTimeout(1, TimeUnit.SECONDS).join();
        display("2 - temperature reached: " + format(measure) + " °C");
    } catch (CompletionException e) {
        display("2 - 30 °C not reached in one second: " + e.getCause().getClass().getSimpleName());
    }
    thermometer.stop();
}

// turns the thermometer's notifications into a CompletableFuture
CompletableFuture<Double> waitForTemperature(Thermometer thermometer, double threshold) {
    // a CompletableFuture that we will complete ourselves
    CompletableFuture<Double> result = new CompletableFuture<>();
    // the listener completes the CompletableFuture when the threshold is reached
    DoubleConsumer listener = temperature -> {
        if (temperature >= threshold) {
            result.complete(temperature);
        }
    };
    thermometer.addListener(listener);
    // whatever the way it ends (threshold, timeout, cancellation), we remove the listener
    result.whenComplete((temperature, e) -> thermometer.removeListener(listener));
    // the CompletableFuture is returned immediately; it will be completed later
    return result;
}

// a thermometer that notifies its listeners every 200 ms (an "old-style" API)
class Thermometer {
    private final List<DoubleConsumer> listeners = new CopyOnWriteArrayList<>();
    private final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();
    private double temperature = 19;

    void addListener(DoubleConsumer listener) {
        listeners.add(listener);
    }

    void removeListener(DoubleConsumer listener) {
        listeners.remove(listener);
    }

    void start() {
        timer.scheduleAtFixedRate(() -> {
            temperature += 0.5;
            IO.println("          measure: " + format(temperature) + " °C");
            listeners.forEach(listener -> listener.accept(temperature));
        }, 200, 200, TimeUnit.MILLISECONDS);
    }

    void stop() {
        timer.shutdown();
    }
}

// a real number in the local format, without useless decimals
String format(double value) {
    return NumberFormat.getInstance().format(value);
}

// display with the time elapsed since the start
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async22.java
          measure: 19.5 °C
          measure: 20 °C
          measure: 20.5 °C
          measure: 21 °C
  807 ms : 1 - temperature reached: 21 °C
          measure: 21.5 °C
          measure: 22 °C
          measure: 22.5 °C
          measure: 23 °C
          measure: 23.5 °C
 1809 ms : 2 - 30 °C not reached in one second: TimeoutException

Comments:

• Lines 39–64: The thermometer, an “old-school” class: a [ScheduledExecutorService] increases the temperature by 0.5 °C every 200 ms and transmits it to its listeners, of type [DoubleConsumer] (a [Consumer] specialized for the [double] type). The list of listeners is a [CopyOnWriteArrayList], a list that can be modified by one thread (adding or removing a listener) while another thread iterates through it;
• lines 22–37: the method that converts notifications into a [CompletableFuture]. It creates a temporary [CompletableFuture<Double>], registers a listener with the thermometer, and immediately returns the [CompletableFuture], which is not yet complete;
• lines 27–31: With each measurement, the listener checks the threshold. When it is reached, it terminates the [CompletableFuture] with the temperature as the result. Subsequent measurements continue to call [complete], to no effect, until the listener is removed;
• line 34: Regardless of how [CompletableFuture] ends (threshold reached, timeout, cancellation by the caller), [whenComplete] removes the listener. Without this, a listener whose threshold is never reached would remain registered indefinitely;
• Lines 8–18: For the caller, waiting for a temperature is an asynchronous operation like any other: the caller waits for it with [join()] and sets a time limit on it with [orTimeout]. Since [orTimeout] terminates [CompletableFuture] itself (and not a copy), the action [whenComplete] on line 34 is executed, and the listener is removed.

A note of caution: the continuations attached to [CompletableFuture] are executed by the thread that calls [complete]—in this case, the thermometer thread—in the middle of notifying its listeners. If these continuations are lengthy, they should be written using the [xxxAsync] variants and an executor, or [CompletableFuture] should be terminated with [completeAsync(() -> value, executor)] (Java 9), which causes the executor to call [complete].

9.11. Limited Parallelism

9.11.1. Virtual threads and semaphores

[allOf] and one virtual thread per operation launch all operations simultaneously. This is not always desirable: to download 10,000 web pages, you don’t want to open 10,000 simultaneous connections to the same server. With platform threads, concurrency was limited by the size of the thread pool. With virtual threads, threads are not pooled (they are so inexpensive that one is created per task): access to the scarce resource is limited using a [Semaphore] (Chapter 8), whose permissions represent the number of concurrent operations allowed. The program [Async23.java] downloads nine pages, no more than three at a time:

// processing a collection with a limited number of simultaneous operations: virtual threads and Semaphore
final AtomicInteger inProgress = new AtomicInteger();

long start;

void main() {
    start = System.nanoTime();
    List<String> pages = IntStream.rangeClosed(1, 9).mapToObj(i -> "page" + i + ".html").toList();
    // at most 3 downloads at the same time: a semaphore with 3 permits
    var permits = new Semaphore(3);
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        // one virtual thread per page: they are cheap, we can create as many as there are pages
        for (String page : pages) {
            executor.submit(() -> {
                // the thread waits for a permit (without blocking a carrier thread)
                permits.acquire();
                try {
                    download(page);
                } finally {
                    permits.release();
                }
                return null;
            });
        }
    } // close() waits for the end of all the downloads
    display("the " + pages.size() + " pages are downloaded");
}

void download(String page) throws InterruptedException {
    // counter shared by the simultaneous operations: we modify it atomically
    int n = inProgress.incrementAndGet();
    display("start of " + page + " (" + n + " in progress)");
    // the simulated download of a page
    Thread.sleep(1000);
    inProgress.decrementAndGet();
    display("end of " + page);
}

// display with the time elapsed since the start and the thread executing the display
void display(String message) {
    Thread thread = Thread.currentThread();
    String name = thread.isVirtual() ? "virtual #" + thread.threadId() : thread.getName();
    IO.println(String.format("%5d ms [%-11s] : %s", (System.nanoTime() - start) / 1_000_000, name, message));
}

Results:

> java Async23.java
   13 ms [virtual #25] : start of page2.html (2 in progress)
   13 ms [virtual #27] : start of page3.html (3 in progress)
   13 ms [virtual #23] : start of page1.html (1 in progress)
 1026 ms [virtual #25] : end of page2.html
 1026 ms [virtual #27] : end of page3.html
 1026 ms [virtual #23] : end of page1.html
 1026 ms [virtual #28] : start of page4.html (1 in progress)
 1027 ms [virtual #29] : start of page5.html (2 in progress)
 1027 ms [virtual #30] : start of page6.html (3 in progress)
 2027 ms [virtual #29] : end of page5.html
 2027 ms [virtual #30] : end of page6.html
 2028 ms [virtual #31] : start of page7.html (1 in progress)
 2028 ms [virtual #32] : start of page8.html (2 in progress)
 2027 ms [virtual #28] : end of page4.html
 2028 ms [virtual #33] : start of page9.html (3 in progress)
 3028 ms [virtual #31] : end of page7.html
 3028 ms [virtual #32] : end of page8.html
 3029 ms [virtual #33] : end of page9.html
 3030 ms [main       ] : the 9 pages are downloaded

Comments:

• line 10: a semaphore with 3 permissions;
• lines 13–24: one virtual thread per page. The task is a lambda that ends with [return null;]: this is therefore a [Callable], whose method can throw checked exceptions, which avoids having to handle the [InterruptedException], [acquire], and [Thread.sleep] here;
• Lines 16–21: Each thread waits for a permit ([acquire]), downloads its page, and then returns the permit ([release]) in a [finally], so that it is returned even if the download fails. A virtual thread waiting for authorization is detached from its host thread, just as during an I/O operation;
• Lines 29–37: The counter [inProgress], shared by the threads, is an atomically modified version of [AtomicInteger];
• The results show three waves of three pages: as soon as one page is finished, another begins, and the number of downloads in progress never exceeds 3. The entire process takes 3 seconds instead of 9. When two threads write at nearly the same time, the order of the lines may vary from one execution to another, and a displayed time may even be slightly earlier than that of the previous line;
• each page has its own virtual thread (nos. 23 through 33; the numbers are shared by all threads in JVM, including the carrier threads, hence the gaps), but these nine virtual threads were executed by the machine’s only two carrier threads.

9.11.2. Producers / Consumers: BlockingQueue

The previous chapter introduced the producer/consumer model: producers place items into a queue, and consumers remove them. The [BlockingQueue<E>] interface is the tool for this: [put] places an item in the queue and waits—if the queue is bounded and full—for a spot to become available; [take] removes an element and waits—if the queue is empty—for one to arrive; [offer] and [poll] attempt the same thing without waiting (or with a delay). These waits are blocking; when executed by virtual threads, they do not block any system threads, and the blocking queue becomes the communication tool between asynchronous operations.

The program [Async24.java] has commands—generated every 50 ms—processed by two consumers, each of which takes 400 ms to process a command:

// producer / consumers with a blocking queue, executed by virtual threads
// the value that signals to the consumers that there will be no more orders
static final int END = -1;
// a bounded queue: it holds at most 2 orders waiting to be processed
final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(2);
long start;

void main() {
    start = System.nanoTime();
    // the producer and two consumers work at the same time
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        executor.submit(() -> produce(8));
        executor.submit(() -> consume("C1"));
        executor.submit(() -> consume("C2"));
    }
    display("all the orders have been processed");
}

// the producer: one order every 50 ms
Void produce(int nbOrders) throws InterruptedException {
    for (int i = 1; i <= nbOrders; i++) {
        Thread.sleep(50);
        // offer puts the order immediately if there is room in the queue
        if (queue.offer(i)) {
            display("P  : order " + i + " queued");
        } else {
            // queue full: put waits for a place to become free
            display("P  : queue full, order " + i + " waits");
            queue.put(i);
            display("P  : order " + i + " queued");
        }
    }
    // nothing more will be produced: we put the end marker
    queue.put(END);
    display("P  : end of production");
    return null;
}

// a consumer: processing an order takes 400 ms
Void consume(String name) throws InterruptedException {
    while (true) {
        // take waits until there is an order in the queue
        int order = queue.take();
        if (order == END) {
            // we put the end marker back in the queue for the other consumer
            queue.put(END);
            break;
        }
        display(name + " : processing of order " + order);
        Thread.sleep(400);
    }
    display(name + " : no more orders");
    return null;
}

// display with the time elapsed since the start
void display(String message) {
    IO.println(String.format("%5d ms : %s", (System.nanoTime() - start) / 1_000_000, message));
}

Results:

> java Async24.java
   58 ms : P  : order 1 queued
   62 ms : C1 : processing of order 1
  118 ms : C2 : processing of order 2
  119 ms : P  : order 2 queued
  169 ms : P  : order 3 queued
  220 ms : P  : order 4 queued
  271 ms : P  : queue full, order 5 waits
  469 ms : P  : order 5 queued
  469 ms : C1 : processing of order 3
  519 ms : C2 : processing of order 4
  519 ms : P  : order 6 queued
  570 ms : P  : queue full, order 7 waits
  870 ms : C1 : processing of order 5
  870 ms : P  : order 7 queued
  920 ms : C2 : processing of order 6
  920 ms : P  : order 8 queued
 1270 ms : C1 : processing of order 7
 1271 ms : P  : end of production
 1320 ms : C2 : processing of order 8
 1672 ms : C1 : no more orders
 1721 ms : C2 : no more orders
 1721 ms : all the orders have been processed

Comments:

• line 5: a [ArrayBlockingQueue] with capacity 2: it contains at most 2 pending commands;
• Lines 11–15: The producer and the two consumers are executed by three virtual threads; the program waits for all three to finish using the executor’s [close()] method. The methods [produce] and [consume] return [Void] and end with [return null;]: the lambdas on lines 12–14 are therefore [Callable], and the methods can declare [InterruptedException];
• Lines 19–37: The producer attempts to submit each order using [offer]. If the queue is full, it signals this and waits for space using [put]: the producer is throttled to the consumers’ pace (lines 8–9 of the results). This is the benefit of a bounded queue: a producer that is too fast cannot fill up memory;
• line 34: production is complete; the producer places a specific value, the end marker [END] (line 3), which the consumers will recognize. A blocking queue, in fact, has no way of signaling that it is “closed”;
• Lines 39–54: Each consumer retrieves orders with [take]. An order is delivered to only one consumer: orders are distributed between C1 and C2 based on their availability. The consumer that receives the end token places it back in the queue for the other consumer, then stops;
• as in the previous example, two lines displayed at nearly the same time may appear out of order.

9.12. Time: ScheduledExecutorService and Clock

Two tools complement those related to time:

• a [ScheduledExecutorService] (Java 5) is an executor capable of running a task after a delay ([schedule]) or periodically ([scheduleAtFixedRate], at fixed intervals measured from the start of one execution to the next; [scheduleWithFixedDelay], with a fixed delay between the end of one execution and the start of the next). A periodic task is never executed twice at the same time: if an execution takes longer than the period, the next one is delayed. A single-threaded executor is obtained with [Executors.newSingleThreadScheduledExecutor()];
• the abstract class [java.time.Clock] (Java 8) represents the clock: it provides the current time and time zone. [Clock.systemDefaultZone()] is the system clock in the machine’s time zone. All [now()] methods of API, [java.time] ([LocalDateTime.now(clock)], [Instant.now(clock)]…) accept a clock. A class that depends on the time receives a [Clock] instead of calling [LocalDateTime.now()]: it can then be tested with a controlled clock.

The program [Async25.java] uses both:

// time: ScheduledExecutorService and Clock
void main() throws InterruptedException {
    // 1 - an action every 500 ms, stopped after 2.2 seconds
    long start = System.nanoTime();
    AtomicInteger nbTicks = new AtomicInteger();
    try (ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor()) {
        // scheduleAtFixedRate: first tick in 500 ms, then one tick every 500 ms
        ScheduledFuture<?> ticks = scheduler.scheduleAtFixedRate(
                () -> IO.println(String.format("%6d ms : tick no. %d",
                        (System.nanoTime() - start) / 1_000_000, nbTicks.incrementAndGet())),
                500, 500, TimeUnit.MILLISECONDS);
        // schedule: a single action, in 2.2 seconds, that stops the ticks
        ScheduledFuture<?> stop = scheduler.schedule(
                () -> ticks.cancel(false), 2200, TimeUnit.MILLISECONDS);
        stop.get();
        IO.println(String.format("%6d ms : timer stopped after %d ticks",
                (System.nanoTime() - start) / 1_000_000, nbTicks.get()));
    } catch (ExecutionException e) {
        IO.println("error: " + e.getCause());
    }

    // 2 - Clock: a class that depends on the time receives its clock instead of calling LocalDateTime.now()
    Clock morning = Clock.fixed(Instant.parse("2026-10-03T09:30:00Z"), ZoneOffset.UTC);
    Clock evening = Clock.fixed(Instant.parse("2026-10-03T19:45:00Z"), ZoneOffset.UTC);
    IO.println(new Reception(morning).greet("Pierre"));
    IO.println(new Reception(evening).greet("Pierre"));
    IO.println(new Reception(Clock.systemDefaultZone()).greet("Pierre") + " (system clock)");
}

// a class that depends on the current time
record Reception(Clock clock) {
    String greet(String name) {
        LocalDateTime now = LocalDateTime.now(clock);
        String greeting = now.getHour() < 18 ? "Hello" : "Good evening";
        String date = now.format(DateTimeFormatter.ofPattern("dd/MM/yyyy HH:mm"));
        return "[" + date + "] " + greeting + " " + name;
    }
}

Results:

> java Async25.java
   504 ms : tick no. 1
  1003 ms : tick no. 2
  1503 ms : tick no. 3
  2003 ms : tick no. 4
  2204 ms : timer stopped after 4 ticks
[03/10/2026 09:30] Hello Pierre
[03/10/2026 19:45] Good evening Pierre
[03/10/2026 19:18] Good evening Pierre (system clock)

Comments:

• Lines 8–11: [scheduleAtFixedRate(task, initialDelay, period, unit)] runs the task for the first time after 500 ms, and then every 500 ms. It returns a [ScheduledFuture] and a [Future], which, among other things, allows future executions to be canceled. The top counter is a [AtomicInteger]: a lambda cannot modify a local variable;
• Lines 13–15: [schedule(task, delay, unit)] executes a single task after 2.2 seconds: it cancels the tops. [stop.get()] waits for this task to finish. The beeps occurred at 0.5, 1, 1.5, and 2 seconds, and the timer stops at 2.2 seconds, after 4 beeps;
• lines 30–38: the record [Reception] says “Hello” or “Good evening” depending on the time. It receives its clock from its constructor instead of calling [LocalDateTime.now()]: this is an injected dependency, just as the [dao] layer is injected into the [business] layer in Chapter 6;
• Lines 23–27: [Clock.fixed(instant, zone)] returns a stopped clock that always shows the same time; this allows us to verify the behavior of [Reception] in the morning and evening without having to wait until evening. The last line uses the system clock (so the result depends on the execution time). For more sophisticated tests, you can write your own class derived from [Clock], which you can advance as desired. The [InstantSource] interface (Java 17), which returns only the current time, is an even lighter dependency when the time zone is irrelevant.

To measure a duration, do not use the clock (which can be adjusted during the measurement, for example, by synchronizing the system time) but use [System.nanoTime()], as shown in the examples in this chapter.

9.13. Structured Concurrency (Overview)

In the previous examples, the operations launched by a program run independently of it: if one fails, the others continue; if the code that launched them is interrupted, they continue as well; and nothing forces that code to wait for them to finish. Structured concurrency proposes treating a group of subtasks as a single block: they are launched within a scope, the scope waits for them to complete, and if one fails, the others are canceled. The [StructuredTaskScope] class implements this idea using virtual threads. It has been available since Java 21 but is still in “preview” status in Java 25: its API may still change, and a program that uses it must be compiled and run with the [--enable-preview] option. It is therefore not included in the course exercises; the following code (file [preview/StructuredConcurrency.txt], to be copied into a file named [StructuredConcurrency.java] to test it) gives an idea of how it works:

// structured concurrency: preview API in Java 25
// execution: java --enable-preview StructuredConcurrency.java
import java.util.concurrent.StructuredTaskScope.Joiner;
import java.util.concurrent.StructuredTaskScope.Subtask;

void main() throws InterruptedException {
    // 1 - two subtasks that must both succeed
    try (var scope = StructuredTaskScope.open()) {
        Subtask<Integer> angers = scope.fork(() -> readTemperature("Angers", 1000));
        Subtask<Integer> brest = scope.fork(() -> readTemperature("Brest", 500));
        // join waits for the end of the subtasks; if one fails, the other is cancelled
        // and join throws an exception
        scope.join();
        IO.println("Angers: " + angers.get() + ", Brest: " + brest.get());
    }
    // 2 - the first answer is enough: the other subtasks are cancelled
    try (var scope = StructuredTaskScope.open(Joiner.<Integer>anySuccessfulResultOrThrow())) {
        scope.fork(() -> readTemperature("mirror-paris", 800));
        scope.fork(() -> readTemperature("mirror-lyon", 300));
        IO.println("first answer: " + scope.join());
    }
}

int readTemperature(String city, long duration) throws InterruptedException {
    Thread.sleep(duration);
    return 10 + city.length();
}

Results:

> java --enable-preview StructuredConcurrency.java
Angers: 16, Brest: 15
first answer: 21

Comments:

• line 8: [StructuredTaskScope.open()] opens a scope, declared in a [try] with a resource; closing it ensures that no subtasks survive it;
• Lines 9–10: [fork] launches a subtask in a new virtual thread and returns a [Subtask];
• line 13: [join()] waits for all subtasks to finish. With the default policy, if a subtask fails, the others are canceled (aborted) and [join()] throws an exception;
• line 14: the results of the subtasks, all of which have completed;
• Lines 17–21: Another policy, provided by [Joiner]: the first successful subtask returns the result of [join()], and the others are canceled. This is the structured equivalent of [anyOf], but the losing requests are actually terminated.

9.14. Application: Tax Calculation with Virtual Threads

The program [Async26.java] brings together several tools from this chapter related to the tax calculation of the main thread. It reads the tax brackets from the file [TaxData.txt], which has already been used, and reads a list of taxpayers from the file [taxpayers.txt]:

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

(each line contains: married [y/n], number of children, annual salary), calculates each person’s tax, and writes the results to the file [results.txt]. Both reads are performed concurrently by virtual threads, and the entire process is limited to 10 seconds.

// application: the tax calculation, with input / output executed by virtual threads
// - the brackets are read in [TaxData.txt], the taxpayers in [taxpayers.txt]
// - the results are written in [results.txt]
// a correct taxpayer line: married (y/n), number of children, annual salary
static final Pattern TAXPAYER = Pattern.compile("([yn])\\s+(\\d+)\\s+(\\d+)");

void main() {
    Path results = Path.of("results.txt");
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        // the two input files are read at the same time, by two virtual threads
        Future<List<TaxBracket>> bracketsReading = executor.submit(
                () -> loadBrackets(Path.of("TaxData.txt")));
        Future<List<String>> taxpayersReading = executor.submit(
                () -> Files.readAllLines(Path.of("taxpayers.txt")));
        // the calculation waits for the two readings, then writes the results
        Future<Integer> processing = executor.submit(
                () -> calculateTaxes(bracketsReading.get(), taxpayersReading.get(), results));
        try {
            // the whole processing is limited to 10 seconds
            int nbResults = processing.get(10, TimeUnit.SECONDS);
            IO.println(nbResults + " lines written in [" + results + "] :");
            IO.print(Files.readString(results));
        } catch (TimeoutException e) {
            processing.cancel(true);
            IO.println("processing abandoned: timeout exceeded");
        }
    } catch (ExecutionException e) {
        // the error of a reading is wrapped twice: by the reading and by the calculation that waited for it
        Throwable cause = e.getCause();
        while (cause instanceof ExecutionException) {
            cause = cause.getCause();
        }
        IO.println("error: " + cause.getClass().getSimpleName() + " : " + cause.getMessage());
    } catch (IOException | InterruptedException e) {
        IO.println("error: " + e.getClass().getSimpleName() + " : " + e.getMessage());
    }
}

// reads the tax brackets (limit:coeffR:coeffN, decimal point)
List<TaxBracket> loadBrackets(Path file) throws IOException {
    try (Stream<String> lines = Files.lines(file)) {
        return lines.filter(line -> !line.isBlank()).map(TaxBracket::parse).toList();
    }
}

// calculates the tax of each taxpayer and writes the results
int calculateTaxes(List<TaxBracket> brackets, List<String> taxpayers, Path output) throws IOException {
    try (BufferedWriter writer = Files.newBufferedWriter(output)) {
        for (String line : taxpayers) {
            Matcher fields = TAXPAYER.matcher(line.trim());
            String result = fields.matches()
                    ? "tax = " + calculateTax(brackets, fields.group(1).equals("y"),
                            Integer.parseInt(fields.group(2)), Integer.parseInt(fields.group(3))) + " €"
                    : "incorrect line";
            writer.write(String.format("%-12s : %s", line, result));
            writer.newLine();
        }
    }
    return taxpayers.size();
}

// the tax calculation (2004 tax scale on 2003 income, kept for the exercise)
int calculateTax(List<TaxBracket> brackets, boolean married, int nbChildren, int salary) {
    BigDecimal shares = BigDecimal.valueOf(nbChildren).divide(BigDecimal.TWO)
            .add(BigDecimal.valueOf(married ? 2 : 1));
    if (nbChildren >= 3) {
        shares = shares.add(new BigDecimal("0.5"));
    }
    BigDecimal income = new BigDecimal("0.72").multiply(BigDecimal.valueOf(salary));
    BigDecimal quotient = income.divide(shares, MathContext.DECIMAL64);
    // the first bracket whose limit exceeds the family quotient, otherwise the last one
    TaxBracket bracket = brackets.stream()
            .filter(t -> quotient.compareTo(t.limit()) <= 0)
            .findFirst()
            .orElse(brackets.getLast());
    return bracket.coeffR().multiply(income).subtract(bracket.coeffN().multiply(shares)).intValue();
}

// a tax bracket
record TaxBracket(BigDecimal limit, BigDecimal coeffR, BigDecimal coeffN) {
    static TaxBracket parse(String line) {
        String[] fields = line.split(":");
        return new TaxBracket(new BigDecimal(fields[0]), new BigDecimal(fields[1]),
                new BigDecimal(fields[2]));
    }
}

Results:

> java Async26.java
7 lines written in [results.txt] :
y 2 60000    : tax = 4282 €
y 3 60000    : tax = 2959 €
n 0 60000    : tax = 11519 €
n 2 25000    : tax = 792 €
y 5 120000   : tax = 11109 €
x 1 20000    : incorrect line
y 1 -5       : incorrect line

The same program, run in a folder where the file [TaxData.txt] is missing:

> java Async26.java
error: NoSuchFileException : TaxData.txt

Comments:

• lines 11–14: The two input files are read simultaneously by two virtual threads using standard blocking read methods ([Files.lines], [Files.readAllLines]);
• lines 16–17: The calculation is a third task, which begins by waiting for the two reads to complete using [get()]. These waits block a virtual thread, which incurs no cost;
• lines 18–26: The main program waits for the result for up to 10 seconds. After that, it aborts the process (which terminates its virtual thread) and reports the error;
• lines 27–36: Errors are handled in a single location within the main program. A read error (missing file, incorrect number) is stored in the [Future] associated with the read operation; the [get()] from the calculation rethrows it into a [ExecutionException], which in turn becomes the calculation’s exception, rethrown by the main program’s [get] into a second [ExecutionException]. The loop in lines 30–32 traces back to the original cause (second result). This is the drawback of dependent tasks; structured concurrency, as described in the previous paragraph, will eliminate this issue;
• lines 39–44: reading the brackets, using a [Stream] that skips empty lines;
• lines 46–60: processing taxpayers. Each line is validated by the regular expression in line 5: a [y] or an [n], followed by two positive integers separated by spaces ([\s+]). The parentheses define three groups, which [group(1)], [group(2)], and [group(3)] return once [matches()] has verified that the entire line matches the expression. An incorrect line ([x] instead of [y/n], negative salary) is flagged as such;
• Lines 62–77: tax calculation, in [BigDecimal]. Dividing the income by the number of shares may not be exact: this is done with a precision of 16 significant digits ([MathContext.DECIMAL64]). The first tax bracket whose threshold is greater than or equal to the family quotient is determined by a stream; [orElse(brackets.getLast())] returns the last bracket if none match. [intValue()] rounds the tax down to the nearest euro;
• the results provide the control values for the red thread: €4,282 for a married couple with 2 children and a salary of €60,000, €2,959 with 3 children, and €11,519 for a single person without children.

The calculation itself is synchronous: it performs no I/O and takes a few microseconds. It would be pointless—and even harmful—to offload it to another thread. Only tasks that involve waiting are made asynchronous.

9.15. The Golden Rules

Let’s summarize the rules of asynchronous programming in Java:

• to wait, do not block a system thread: a wait (I/O, timeout, or other operation) is performed either in a virtual thread or by attaching a continuation to a [CompletableFuture];
• for new code, prefer virtual threads: write blocking, sequential, and readable code (loops, [try / catch], ordinary local variables), executed by [Executors.newVirtualThreadPerTaskExecutor()]; one virtual thread per task, never use a pool of virtual threads;
• Use [CompletableFuture] to compose operations, or to use a API that returns [CompletableFuture] ([HttpClient.sendAsync], reactive libraries); We then choose the continuation executor ([xxxAsync(..., executor)]) and never execute a blocking operation in the shared pool;
• no [join()] or [get()] in a continuation or in a GUI thread: these waits block the thread executing them. They belong in the main program or in a virtual thread;
• Virtual threads are used for waits, not computations: for computations, use a platform-specific thread pool sized according to the number of processors, or parallel streams;
• limit access to scarce resources using a [Semaphore], not by the size of a thread pool;
• Cancellation is cooperative: a task is canceled via [Future.cancel(true)], which interrupts its thread; an interrupted task terminates gracefully; a [InterruptedException] is never masked (it is either propagated or the interrupt flag is reset). And remember that [CompletableFuture.cancel] does not interrupt anything;
• limit wait times: [orTimeout], [completeOnTimeout], [get(timeout, unit)]; and terminate the operation itself if the wait is abandoned;
• start first, then wait: to run multiple independent operations simultaneously, start them all and then wait for them ([allOf], [invokeAll], the executor’s [close] method);
• A launched operation must be waited on or monitored: a forgotten operation hides its exceptions ([exceptionally], [whenComplete] to handle them); by tracing back a [CompletionException] or a [ExecutionException], we seek the root cause.

The following table summarizes the equivalences between the world of platform threads and blocking waits (previous chapter), that of [CompletableFuture], and that of virtual threads:

Platform threads, blocking waits	CompletableFuture	Virtual threads
[Thread.sleep(1000)] (blocks the thread)	[new CompletableFuture<Void>().completeOnTimeout(null, 1, SECONDS)]	[Thread.sleep(1000)] (releases the carrier)
[thread.start()] followed by [thread.join()]	[supplyAsync(...)] followed by [join()] or a continuation	[executor.submit(...)] followed by [get()]
Wait for multiple threads ([join] in a loop)	[allOf(...)], [anyOf(...)]	[invokeAll(...)], [ExecutorCompletionService], [close()]
Recall methods ([CompletionHandler])	[thenApply], [thenCompose], [thenCombine]	The code following the blocking call
[try / catch] around [get()] ([ExecutionException])	[exceptionally], [handle], [whenComplete]	[try / catch] (standard)
[thread.interrupt()]	[cancel(...)] (does not interrupt anything)	[future.cancel(true)] (interrupts the thread)
[future.get(timeout, unit)]	[orTimeout(...)], [completeOnTimeout(...)]	[future.get(timeout, unit)]
fixed-size thread pool	continuation executor [xxxAsync]	[Semaphore]
[BlockingQueue] (blocks the thread)	[Flow], [SubmissionPublisher]	[BlockingQueue] (releases the carrier)
[ScheduledExecutorService]	[completeOnTimeout], [delayedExecutor]	[Thread.sleep] in a loop

The following chapters will apply these rules: database access using JDBC (Chapter 10) and Hibernate (Chapter 11), whose API operations are blocking and work naturally with virtual threads, the TCP server, which serves each client via a virtual thread, and the [HttpClient] client (Chapter 12); Spring Boot web services, where the Tomcat server can process each request via a virtual thread; and their clients (Chapter 13).