CompletableFuture, and Future

1. Overview

CompletableFuture and Future are used for asynchronous programming in Java. Asynchronous programming is a means of writing non-blocking code by running a task on a separate thread than the main application thread and notifying the main thread about its progress, completion or failure. So Main thread doesn't block/wait for the completion of an async task and it can execute other tasks in parallel.

 

• Future's drawbacks
  
  • Cannot perform further action on a Future's result without blocking
  • Cannot combine multiple Futures together
  • Cannot be chained together
  • No Exception Handling

 

CompletableFuture implements the Future interface so can use it as a Future implementation, but with additional completion logic. It gets over all of Future's drawbacks with over 50 different methods for composing, combining, executing asynchronous computation steps and handling errors.

 

2. Usages

• Trivial

Creating CompletableFuture instance with no-arg is done like below.

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

To get a result of this instance is to call get() method like below. but this approach blocks thread til Future is complete.

String result = completableFuture.get()

You can escape this blocking to call complete() method like below

public static <T> CompletableFuture<T> firstOrNull(List<CompletableFuture<T>> futures, Predicate<T> condition) {
    CompletableFuture<T> finalResult = new CompletableFuture<>();
    // attempt to complete on success
    futures.stream().forEach(future -> future.thenAccept(successResult -> {
        if (condition.test(successResult))
            finalResult.complete(successResult);
    }));
    CompletableFuture<?> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
    all.thenRun(() -> {
    	// this method ignored if future has already been completed.
        finalResult.complete(null);
    });
    return finalResult;
}

• runAsync()

if wanting to run some background task asynchronously and don't want to return anything from the task, then you can use CompletableFuture.runAsync() method. it takes a Runnable Object and returns completable Future<void>.

// Run a task specified by a Runnable Object asynchronously.
CompletableFuture<Void> future = CompletableFuture.runAsync(new Runnable() {
    @Override
    public void run() {
        // Simulate a long-running Job
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        System.out.println("I'll run in a separate thread than the main thread.");
    }
});

// Block and wait for the future to complete
future.get()

Also passing the Runnable object in the form of a lambda below

// Using Lambda Expression
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // Simulate a long-running Job   
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    System.out.println("I'll run in a separate thread than the main thread.");
});

 

• supplyAsync()

it takes a Supplier<T> and returns CompletableFuture<T> where T is the type of the value obtained by calling the given supplier.

// Run a task specified by a Supplier object asynchronously
CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Result of the asynchronous computation";
    }
});

// Block and get the result of the Future
String result = future.get();
System.out.println(result);

also using lambda expression is possible like below

// Using Lambda Expression
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "Result of the asynchronous computation";
});

 

• Note: Association with Executor and Thread Pool

runAsync() and supplyAsync() are executing their tasks in a separate thread. and these threads are obtained from the global ForkJoinPool.commonPool().

But also creating a ThreadPool and passing it to runAsync() and supplyAsyc() is possible and they execute their task s in a thread obtained from passing Thread pool.

All the methods in the CompletableFuture API has two variants - One which accepts an Executor as an arg and one which doesn't 

// Variations of runAsync() and supplyAsync() methods
static CompletableFuture<Void>  runAsync(Runnable runnable)
static CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

Here's example to pass user creating thread pool 

Executor executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "Result of the asynchronous computation";
}, executor); // passing user thread pool

 

• Callback

CompletableFuture.get() is blocking until the Future is completed and returns the result after its completion. But that's not the asynchronous way to work out. in asynchronous systems, we should be able to attach a callback to the CompletableFuture which should automatically get called when the Future completes.

This way, no need to wait for the result, and then writing the logic that needs to be executed after the completion of the Future inside our callback function using thenApply(), thenAccept(), and thenRun().

 

 

• thenApply()

this method takes a Function<T,R> as an arg. Function<T,R> is a simple functional interface representing a funciton that accepts an arg of type T and produce a result of type R

// Create a CompletableFuture
CompletableFuture<String> whatsYourNameFuture = CompletableFuture.supplyAsync(() -> {
   try {
       TimeUnit.SECONDS.sleep(1);
   } catch (InterruptedException e) {
       throw new IllegalStateException(e);
   }
   return "Rajeev";
});

// Attach a callback to the Future using thenApply()
CompletableFuture<String> greetingFuture = whatsYourNameFuture.thenApply(name -> {
   return "Hello " + name;
});

// Block and get the result of the future.
System.out.println(greetingFuture.get()); // Hello Rajeev

and also Channing thenApply() is possible like below

CompletableFuture<String> welcomeText = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
       throw new IllegalStateException(e);
    }
    return "Rajeev";
}).thenApply(name -> {
    return "Hello " + name;
}).thenApply(greeting -> {
    return greeting + ", Welcome to the CalliCoder Blog";
});

System.out.println(welcomeText.get());
// Prints - Hello Rajeev, Welcome to the CalliCoder Blog

 

3. References

https://www.callicoder.com/java-8-completablefuture-tutorial/