Concurrency Made Simple: Harnessing CompletableFuture#

What Is the Executor Framework?#

The Executor framework abstracts the execution of tasks from their creation. Instead of manually creating and managing threads, you submit tasks (often as Runnable or Callable instances) to an Executor or ExecutorService. This approach allows:

• Task scheduling independent of application logic.
• Reusability of threads across multiple tasks, reducing overhead.
• Centralized management of concurrency rules (queue size, policies, etc.).

Common Types of Executors#

Submitting Tasks to an Executor#

1
ExecutorService executorService = Executors.newFixedThreadPool(4);
2

3
// Submitting a runnable task:
4
executorService.submit(() -> {
5
    System.out.println("Running in a separate thread");
6
});
7

8
// Submitting a callable task:
9
Future<Integer> futureResult = executorService.submit(() -> {
10
    // Some computation
11
    return 42;
12
});

While Future lets you wait for a result and handle potential timeouts, it lacks a more flexible approach for chaining tasks or responding to completion. That’s where CompletableFuture enters the picture.

Classic Future#

A Future<T> represents the eventual result of an asynchronous computation, but it has significant limitations:

• You must call future.get() to retrieve the result, which can block the thread.
• There’s no direct way to chain additional computations once the result is available.
• Handling errors might require additional checks after calling future.get().
• Canceling and avoiding blocking can be awkward unless carefully designed.

CompletableFuture#

A CompletableFuture<T> provides all the functionality of a Future<T> and goes further:

1. Completion callbacks: You can register functions or consumers to be run once the computation completes (with or without a result).
2. Chaining: You can seamlessly chain dependent tasks, letting them execute once the previous stage finishes.
3. Exception handling: Error handling can be integrated into the chain, without traditional try-catch blocks in the main thread.
4. Manual completion: You can programmatically complete or fail the future outside of the normal execution path.

The CompletableFuture class aligns well with modern asynchronous paradigms, reducing boilerplate while improving readability.

Creating a Basic CompletableFuture#

The simplest way to create a CompletableFuture is via CompletableFuture.supplyAsync(...) or CompletableFuture.runAsync(...).

1
// supplyAsync returns a result
2
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
3
    // Simulate a long-running computation
4
    try {
5
        Thread.sleep(1000);
6
    } catch (InterruptedException e) {
7
        Thread.currentThread().interrupt();
8
    }
9
    return 42;
10
});
11

12
// runAsync does not return a result
13
CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> {
14
    // Some side effect or void-returning work
15
    System.out.println("Running async task");
16
});

Both methods accept an optional Executor parameter if you don’t want to rely on the default ForkJoinPool.commonPool().

Retrieving the Result#

To retrieve the result from a CompletableFuture<T>, you can still use the get() method:

1
try {
2
    Integer result = future.get(); // Blocks until the result is available
3
    System.out.println("Result: " + result);
4
} catch (InterruptedException | ExecutionException e) {
5
    e.printStackTrace();
6
}

However, blocking with get() is not always optimal since it ties up a thread. Instead, you can harness non-blocking callbacks or chaining methods, which we will explore in upcoming sections.

Completing a CompletableFuture Manually#

One of the unique strengths of CompletableFuture is that it can be manually completed:

1
CompletableFuture<String> manualFuture = new CompletableFuture<>();
2

3
// In some thread, you can do:
4
manualFuture.complete("Manually completed value");
5

6
// If an error occurs, you can also do:
7
manualFuture.completeExceptionally(new RuntimeException("Oops!"));

This feature is particularly useful when integrating with event-driven systems or callback-based APIs. You can create a CompletableFuture, pass a listener to some asynchronous API, and call complete(...) or completeExceptionally(...) once you receive the callback notification.

thenApply and thenAccept#

When you have a CompletableFuture<T>, you can attach a function that transforms the result into a different type:

1
CompletableFuture<String> futureString = future.thenApply(result -> {
2
    return "Result was: " + result;
3
});

Here, future is a CompletableFuture<Integer> from the previous example, so futureString becomes a CompletableFuture<String>.

Alternatively, if you just want to process the result without returning a new value, use thenAccept():

1
future.thenAccept(result -> {
2
    System.out.println("The result is " + result);
3
});

thenCompose for Sequential Steps#

Use thenCompose() when the next step in your chain returns another CompletableFuture:

1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);
2
CompletableFuture<Integer> future2 = future1.thenCompose(value ->
3
    CompletableFuture.supplyAsync(() -> value * 2)
4
);
5
// future2 will complete with 20

thenCompose() flattens nested CompletableFuture<CompletableFuture<U>> into a single CompletableFuture<U>, making it straightforward to sequence asynchronous tasks.

thenCombine#

Imagine you have two independent tasks whose results you want to combine:

1
CompletableFuture<Integer> futureA = CompletableFuture.supplyAsync(() -> 10);
2
CompletableFuture<Integer> futureB = CompletableFuture.supplyAsync(() -> 5);
3

4
CompletableFuture<Integer> combined = futureA.thenCombine(futureB, (a, b) -> a + b);
5
// combined will complete with 15

thenAcceptBoth#

If you simply want to consume both results without returning anything:

1
futureA.thenAcceptBoth(futureB, (a, b) -> {
2
    System.out.println("Sum is " + (a + b));
3
});

allOf and anyOf#

Sometimes, you might have multiple futures in a collection and need to wait for all of them or any one of them to complete.

1
CompletableFuture<Void> allFutures = CompletableFuture.allOf(futureA, futureB);
2
// This returns a CompletableFuture<Void> which completes once *all* are done.
3

4
CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(futureA, futureB);
5
// This completes once any of the futures is done, returning that result.

When you use allOf, be mindful that the returned future is of type CompletableFuture<Void>, so you’ll need to individually retrieve results from each future if you need them:

1
CompletableFuture<Void> allDone = CompletableFuture.allOf(futureA, futureB);
2
allDone.thenRun(() -> {
3
    // Retrieve the values
4
    try {
5
        Integer valA = futureA.get();
6
        Integer valB = futureB.get();
7
        System.out.println("All done: " + (valA + valB));
8
    } catch (Exception e) {
9
        e.printStackTrace();
10
    }
11
});

handle()#

The handle() method lets you process both the result and the exception (if any). For example:

1
CompletableFuture<Integer> handled = future.handle((res, ex) -> {
2
    if (ex != null) {
3
        System.out.println("Exception: " + ex.getMessage());
4
        // Fallback
5
        return 0;
6
    } else {
7
        return res * 2;
8
    }
9
});

exceptionally()#

If you only need to handle an exception and still want to continue with the normal chain:

1
CompletableFuture<Integer> recovered = future.exceptionally(ex -> {
2
    System.out.println("Error: " + ex.getMessage());
3
    return -1; // Return a default or fallback value
4
});

whenComplete()#

If you want to handle an exception but still pass the original result (or exception) along, use whenComplete():

1
future.whenComplete((res, ex) -> {
2
    if (ex != null) {
3
        System.out.println("Error: " + ex.getMessage());
4
    } else {
5
        System.out.println("Success: " + res);
6
    }
7
});

whenComplete() allows you to do some logging or cleansing but does not override the result unless you pair it with a transform.

Handling Timeouts#

You can handle timeouts by using the orTimeout and completeOnTimeout methods (introduced in Java 9). For instance:

1
CompletableFuture<Integer> futureWithTimeout = CompletableFuture.supplyAsync(() -> {
2
    try {
3
        Thread.sleep(3000); // Simulating a long task
4
    } catch (InterruptedException e) {
5
        Thread.currentThread().interrupt();
6
    }
7
    return 42;
8
}).orTimeout(2, TimeUnit.SECONDS)
9
  .exceptionally(ex -> {
10
      System.out.println("Timed out!");
11
      return -1;
12
  });

Alternatively, you can manually manage timeouts by using get(long timeout, TimeUnit unit):

1
try {
2
    Integer value = future.get(2, TimeUnit.SECONDS);
3
} catch (TimeoutException e) {
4
    System.out.println("Operation timed out");
5
}

Custom Executors#

By default, static methods like supplyAsync use the ForkJoinPool.commonPool(). In production, you’ll often delegate to a specific ExecutorService:

1
ExecutorService customExecutor = Executors.newFixedThreadPool(4);
2

3
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
4
    // Some CPU or I/O intensive task
5
    return 42;
6
}, customExecutor);

This helps you control the concurrency level, manage thread lifecycles, and isolate different parts of your application to avoid resource contention.

Delayed Execution and Scheduling#

While you can combine CompletableFuture with ScheduledThreadPoolExecutor for delayed execution, CompletableFuture itself doesn’t directly provide scheduling. You can do something like:

1
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
2

3
CompletableFuture<Integer> delayedFuture = new CompletableFuture<>();
4
scheduler.schedule(() -> {
5
    delayedFuture.complete(100);
6
}, 5, TimeUnit.SECONDS);

If you need repeated, periodic tasks, rely on ScheduledExecutorService or frameworks like Spring’s scheduling support combined with CompletableFuture.

Workload Type#

• CPU-bound tasks: If tasks are CPU-intensive, a bounded thread pool (e.g., newFixedThreadPool) is usually best. Creating too many threads can degrade performance.
• I/O-bound tasks: If tasks mostly wait for I/O, a larger thread pool may be appropriate, but still consider non-blocking I/O where possible.

Avoid Oversubscription#

Excess threads can lead to context switching overhead. Instead of arbitrarily scaling thread pools, understand your runtime environment (e.g., the number of available CPU cores, your system load, etc.) to avoid negative performance impacts.

Completion Overhead#

While CompletableFuture is highly optimized, each callback or chained stage adds overhead in memory usage and scheduling. For extremely large chains, consider other patterns like batch processing or more explicit concurrency mechanisms.

Context Propagation#

In many enterprise applications, you need to carry context (security credentials, transaction IDs, user locales, etc.) across threads. Libraries like MicroProfile Context Propagation or custom solutions can help:

• ThreadLocal pitfalls: If you rely on ThreadLocal, you must ensure it’s properly cleared or copied for each asynchronous task.
• Explicit context passing: Consider including context data in the parameters to your methods or hooking into your Executor to set up the context before running the task.

Reactive Bridges#

If you work with frameworks that use reactive programming (e.g., Reactor, RxJava), you can bridge calls to and from CompletableFuture. For instance, in Reactor:

1
Mono<Integer> mono = Mono.fromFuture(() -> future);
2
CompletableFuture<Integer> cf = mono.toFuture();

Such a bridge allows you to combine the expressive power of reactive streams (backpressure, transformations) with the flexibility of CompletableFuture.

Structured Concurrency#

Structured concurrency is a paradigm where concurrency is treated more like structured programming. Popular in some languages like Kotlin (via coroutines), the idea can be implemented in Java with the right patterns:

1. Scoped tasks: Create and manage CompletableFutures in a confined scope (e.g., a method or block) to ensure they are not leaked.
2. Cancellation: Provide mechanisms to cancel running tasks cleanly if a higher-level operation fails or completes early.
3. Error propagation: Make sure that errors bubble up to a well-defined boundary, allowing you to handle them uniformly.

Circuit Breakers, Retrying, and Bulkheading#

In microservices architectures, concurrency patterns often pair with resilience patterns like circuit breakers and retry logic. Libraries like resilience4j can be integrated with CompletableFuture to:

• Circuit Breaker: Open a circuit if a downstream service calling code experiences too many failures.
• Bulkhead: Limit concurrent calls to protect resources from saturation.
• Retries: Attempt re-invocations upon failure, combined with a backoff strategy.

Monitoring and Instrumentation#

Robust production systems require insight into concurrency behaviors:

• Metrics: Track average completion times, success rates, and queue lengths.
• Tracing: Use distributed tracing solutions (e.g., OpenTelemetry) to track asynchronous calls across service boundaries.
• Logging: Use structured logging to see how tasks transition through completion stages.

By incorporating these considerations, you transform CompletableFuture from a simple concurrency tool into a building block for highly resilient and scalable systems.