CompletableFuture is part of the Java java.util.concurrent package, introduced in Java 8. It represents a promise-like construct for asynchronous programming, allowing you to write non-blocking and more readable code for handling computations that may complete in the future.
Life cycle
stateDiagram-v2
[*] --> NotStarted: CompletableFuture created
NotStarted --> Running: Task submitted
Running --> Completed: Task finishes successfully
Running --> ExceptionallyCompleted: Exception occurs
Running --> Canceled: Future canceled
Completed --> [*]: Result available
ExceptionallyCompleted --> [*]: Exception handled
Canceled --> [*]: No result
Completed: Result can be retrieved with get() or join()
ExceptionallyCompleted: Exception can be handled with exceptionally() or handle()
Canceled: Cancel task with cancel(true)
Key Concepts
Promise of a Result:
A CompletableFuture represents a computation that will be completed (successfully or with an exception) at some point.
Non-blocking Execution:
It allows asynchronous computation without blocking the current thread.
Chaining:
You can chain multiple tasks together using methods like .thenApply, .thenAccept, .thenCompose, etc.
Handling Results:
It offers methods to handle the result of the computation, like .get(), .join(), or callbacks for async handling.
Completion:
A CompletableFuture can be completed programmatically using complete() or completeExceptionally().
---
Creating and Using CompletableFuture
Creating a CompletableFuture
create CompletableFuture
Manually created future:
val future = CompletableFuture<String>()// Completing it programmaticallyfuture.complete("Result-Val")println("Some other work")println(future.get())
Output:
Some other workResult-Val
GT-snapshot
Command to reproduce:
gt.sandbox.checkout.commit b581f798c55d205c6e91 \&& cd "${GT_SANDBOX_REPO}" \&& cmd.run.announce "./gradlew run"
val future = CompletableFuture.supplyAsync { out.println("Starting to sleep") Thread.sleep(1000) out.println("Finished sleeping") "Hello, CompletableFuture!"}out.println("Hello, from Main thread")out.println(future.get()) // Output: Hello, CompletableFuture!
GT-Snapshot
Command to reproduce:
gt.sandbox.checkout.commit 9dd9ad6c4750da8e7b93 \&& cd "${GT_SANDBOX_REPO}" \&& cmd.run.announce "./gradlew run"
Recorded output of command:
[2024-11-22T04:33:14.166261Z][elapsed-since-start: 42ms][tname:ForkJoinPool.commonPool-worker-1/tid:20] Starting to sleep[2024-11-22T04:33:14.166259Z][elapsed-since-start: 42ms][tname:main/tid:1] Hello, from Main thread[2024-11-22T04:33:15.193044Z][elapsed-since-start: 1058ms][tname:ForkJoinPool.commonPool-worker-1/tid:20] Finished sleeping[2024-11-22T04:33:15.194765Z][elapsed-since-start: 1059ms][tname:main/tid:1] Hello, CompletableFuture!
When you call thenApply, a new CompletableFuture is created. This new future holds the transformed result of the computation once the original future completes.
The Original Future Remains Unchanged:
The result of the original CompletableFuture (future.get() in your code) is unaffected by the thenApply call. It still holds the value "Jon Snow".
Accessing the Transformed Result:
To retrieve the transformed result (e.g., "Hello Jon Snow" in your example), you need to call .get() or .join() on the CompletableFuture returned by thenApply.
The behavior of thenAccept running on different threads in your code depends on how tasks are scheduled and executed in CompletableFuture. Here’s an explanation:
High-Level Explanation
Thread Pool for supplyAsync: CompletableFuture.supplyAsync by default uses the ForkJoinPool.commonPool, unless you specify a custom Executor. This means any asynchronous task (like supplyAsync) is picked up by one of the threads in the common pool.
Thread Propagation through Chaining:
If a stage in the chain (like handle, thenApply, thenAccept) does not explicitly specify an executor, it may execute:
On the thread completing the previous stage if it happens to be free.
On the calling thread if the task is already completed when the next stage is invoked (i.e., the main thread in this case).
Why thenAccept Runs on Different Threads:
When you print within supplyAsync, it likely adds enough delay for the task to fully complete asynchronously before thenAccept is registered. As a result, thenAccept executes on the ForkJoinPool.commonPool.
When you skip printing, the completion happens quickly enough that the main thread can take over execution of subsequent stages like thenApply and thenAccept.
Recorded Output Analysis
Case 1: When supplyAsync Prints
Thread Behavior:
supplyAsync: Runs on ForkJoinPool.commonPool-worker-1.
thenApply and thenAccept: Also run on the same worker thread since the asynchronous computation was still active in the pool when thenAccept was registered.
Reasoning:
Adding the out.println("supplyAsync") adds latency, keeping the computation on the worker thread longer.
Case 2: When supplyAsync Does Not Print
Thread Behavior:
thenApply and thenAccept: Run on the main thread (main/tid:1).
Reasoning:
Without the delay introduced by out.println, the supplyAsync task completes faster. When the chained operations like thenApply and thenAccept are registered, they are picked up by the thread initiating the chain (the main thread), since the completion occurs synchronously.
Takeaways
Thread Dependence: The thread running thenAccept (or other chained stages) depends on:
Whether the task completed asynchronously or synchronously.
How and when subsequent stages are registered.
Consistency with Custom Executor:
If you want predictable thread execution, supply an Executor to CompletableFuture methods:
ForkJoinPool Behavior: The ForkJoinPool.commonPool can sometimes reuse the same thread for subsequent tasks to optimize throughput, but it’s not guaranteed. Timing and task scheduling influence this behavior.
Solution for Controlling Execution Context
To ensure all stages consistently run on a specific thread pool:
val executor = Executors.newFixedThreadPool(4)CompletableFuture.supplyAsync({ out.println("supplyAsync") "Result"}, executor).thenApplyAsync({ it + " Processed"}, executor).thenAcceptAsync({ out.println(it)}, executor)
Cancellation does NOT work well with CompletableFutures
Cancellation does NOT work well with CompletableFutures
CompletableFuture.cancel
CompletableFuture.cancel(boolean mayInterruptIfRunning). mayInterruptIfRunning: does Nothing.
[Cancellation does NOT work well with CompletableFutures](thorg://notes/u7ox6ra9bc6qdn3zoh7plfm)
```java
/**
* If not already completed, completes this CompletableFuture with
* a {@link CancellationException}. Dependent CompletableFutures
* that have not already completed will also complete
* exceptionally, with a {@link CompletionException} caused by
* this {@code CancellationException}.
*
* @param mayInterruptIfRunning this value has no effect in this
* implementation because interrupts are not used to control
* processing.
*
* @return {@code true} if this task is now cancelled
*/
public boolean cancel(boolean mayInterruptIfRunning) {
```
After 'cancel' is called. Thread.sleep() in cancelled thread does NOT throw Interrupted exception.
Code
package com.glassthought.sandboximport gt.sandbox.util.output.Outimport gt.sandbox.util.output.impl.OutSettingsimport java.util.concurrent.CompletableFutureval out = Out.standard(outSettings = OutSettings(printColorPerThread = true))fun main() { // Create a CompletableFuture that simulates a long-running task val future = CompletableFuture.supplyAsync { out.println("supplyAsync: Task started") try { out.println("Starting 5 second sleep") Thread.sleep(5000) // Simulate a long-running task out.println("Slept for 5 seconds, will sleep for just 100ms more") Thread.sleep(100) out.println("supplyAsync: Task completed") "Task completed" } catch (e: InterruptedException) { out.println("Task was interrupted") throw RuntimeException("Task interrupted", e) } } Thread.sleep(100) // Simulate cancellation after 2 seconds Thread { out.println("Starting a new thread to cancel the future...") Thread.sleep(2000) // Wait 2 seconds before canceling out.println("Cancelling the future...") val cancelled = future.cancel(true) out.println("Future cancelled: $cancelled") }.start() // Attempt to retrieve the result (blocks until completed or cancelled) try { val result = future.get() out.println("Future result: $result") } catch (e: Exception) { out.println("Exception occurred: ${e.message}") } Thread.sleep(6000) out.println("Main thread ends")}
Command to reproduce:
gt.sandbox.checkout.commit e04a4d99c46dad07bed4 \&& cd "${GT_SANDBOX_REPO}" \&& cmd.run.announce "./gradlew run --quiet"
Recorded output of command:
[2024-11-22T16:45:52.824796Z][elapsed-since-start: 15ms][tname:ForkJoinPool.commonPool-worker-1/tid:20] supplyAsync: Task started[2024-11-22T16:45:52.852971Z][elapsed-since-start: 33ms][tname:ForkJoinPool.commonPool-worker-1/tid:20] Starting 5 second sleep[2024-11-22T16:45:52.928483Z][elapsed-since-start: 108ms][tname:Thread-0/tid:21] Starting a new thread to cancel the future...[2024-11-22T16:45:54.931070Z][elapsed-since-start: 2111ms][tname:Thread-0/tid:21] Cancelling the future...[2024-11-22T16:45:54.935170Z][elapsed-since-start: 2115ms][tname:main/tid:1] Exception occurred: null[2024-11-22T16:45:54.938928Z][elapsed-since-start: 2118ms][tname:Thread-0/tid:21] Future cancelled: true[2024-11-22T16:45:57.857226Z][elapsed-since-start: 5037ms][tname:ForkJoinPool.commonPool-worker-1/tid:20] Slept for 5 seconds, will sleep for just 100ms more[2024-11-22T16:45:57.958440Z][elapsed-since-start: 5138ms][tname:ForkJoinPool.commonPool-worker-1/tid:20] supplyAsync: Task completed[2024-11-22T16:46:00.940212Z][elapsed-since-start: 8120ms][tname:main/tid:1] Main thread ends