CompletionStage - Java Async Programming
Future : A Future represents the result of an asynchronous computation.
Future provides below useful methods –
- get() – get is used to get the result of the asynchronous computation.
If computation is still in progress then get() method will block the program in which the get() is called.
So get should be used with proper thought otherwise execution may block. - isDone() – isDone() does not block. Simple returns true if thread is executed and result is available.
In program isDone() can be used to check if asynchronous computation is done and then get()call can b made. - Cancel() – cancellation of the task is done using the cancel method.
As part of the below example we have a Future object which holds the result returned from the thread.
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class MainClass_Callable {
public static void main(String[] args) {
ExecutorCallable callableObj = new ExecutorCallable();
ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
Future<Integer> intResult = executor.submit(callableObj);
try {
if (!intResult.isDone()) {
Thread.sleep(1000);
}
int result = intResult.get();
System.out.println("result from future = " + result);
executor.shutdown();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
Limitation of Future :
- Future object’s get() method blocks the execution, trying to get the thread’s output blocks
further processing. - We cannot chain and combine multiple futures.
- Does not provide exception handling
CompletableFuture:
CompletableFuture adds to the Asynchronous computation capability of the Java that was powered
by Future earlier.
CompletableFuture implements Future and CompletionStage both interfaces. CompletableFuture is a Future
with many other capabilities.
CompletionStage interface consists of many methods for chaining and combining the computation results
as well as to handle
the exceptions.
Some of the frequently used methods from CompletionStage:
- thenApply() : Supplies the result of the completion stage as input and returns a new CompletionStage
that holds the result of the computation passed inside the function.
This method wraps the final output inside a CompletionStage. - thenAccept(): Accept the result of the CompletionStage, and computation passed inside the method completes,
then return a CompletionStage of void type. - thenCombine() : Combine the execution of two different completion stages.
- thenCompose () : Supplies the result of the completion stage as input and returns a new CompletionStage
that holds the result of the computation passed inside the function. - exceptionally() : Whenever there is any exception inside the completion stage or completion stage chain,
program control will come inside this method. This method will be executed whenever there is any exception
and the statement written inside this block will be executed. - handle() : Receives the result of the completion stage and exception if any.
- toCompletableFuture(): Util method to convert from CompletionStage to CompletableFuture.
Executing Completable Futures in serial : Execute all CompletableFuture in Serial
package completables;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
public class CompletableFuturesSerialExecution_All {
public static void main(String[] args) throws InterruptedException, ExecutionException {
CompletableFuturesSerialExecution_All completableFuturesSerialExecution
= new CompletableFuturesSerialExecution_All();
completableFuturesSerialExecution.executeCompletableFuturesSerially();
}
private void executeCompletableFuturesSerially() throws InterruptedException, ExecutionException {
TaskImpl customRuleNumber = new TaskImpl();
for (int i = 0; i < 5; i++) {
CompletableFuture<String> completableFuture = CompletableFuture.completedFuture(null);
completableFuture = executeRule(completableFuture, i, customRuleNumber);
}
}
private CompletableFuture<String> executeRule(CompletableFuture<String> completableFuture,
int i, TaskImpl customRuleNumber)
throws InterruptedException, ExecutionException {
return completableFuture.thenCompose(__ -> customRuleNumber.executeRule(i));
}
abstract class Task {
abstract CompletionStage<String> executeRule(int n);
}
class TaskImpl extends Task {
CompletableFuture<String> executeRule(int n) {
System.out.println("Executing Task number = " + n);
return CompletableFuture.completedFuture("" + n);
};
}
}
Execution Output:
Executing Task number = 0
Executing Task number = 1
Executing Task number = 2
Executing Task number = 3
Executing Task number = 4
Here in the above example, we are executing multiple tasks in the completable futures in
serial order.
Executing Completable Futures in serial : return after one of the futures return the response
package completables;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
public class CompletableFutureSerial_breakAfterResult {
public static void main(String[] args) throws InterruptedException, ExecutionException {
CompletableFutureSerial_breakAfterResult completableFuturesSerialExecution
= new CompletableFutureSerial_breakAfterResult();
completableFuturesSerialExecution.executeCompletableFuturesSerially();
}
private void executeCompletableFuturesSerially() throws InterruptedException, ExecutionException {
TaskImpl customRuleNumber = new TaskImpl();
for (int i = 0; i < 5; i++) {
CompletableFuture<String> completableFuture = CompletableFuture.completedFuture(null);
completableFuture = executeRule(completableFuture, i, customRuleNumber);
if (completableFuture.get() != null) {
System.out.println("Task executed and returned non-null result : Task Number = "
+ completableFuture.get());
break;
}
}
}
private CompletableFuture<String> executeRule(CompletableFuture<String> completableFuture,
int i, TaskImpl customRuleNumber)
throws InterruptedException, ExecutionException {
return completableFuture.thenCompose(__ -> customRuleNumber.executeRule(i));
}
abstract class Task {
abstract CompletionStage<String> executeRule(int n);
}
class TaskImpl extends Task {
CompletableFuture<String> executeRule(int n) {
if (n == 1) {
System.out.println("Executing Task number = " + n);
return CompletableFuture.completedFuture("" + n);
}
System.out.println("returns null");
return CompletableFuture.completedFuture(null);
};
}
}
Output:
returns null
Executing Task number = 1
Task executed and returned non-null result : Task Number = 1
Here in the above example Tas- 0 returns null , then execution continues and when task-1 return the
non-null object then we stop the execution as we have got our result.
We can put a different condition according to our use case and break whenever we want to get out of
the serial execution.
Executing Completable Futures in Parallel : return after all of the futures complete
package completables;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
public class CompletableFuture_Parallel {
public static void main(String[] args) throws InterruptedException, ExecutionException {
CompletableFuture_Parallel completableFuturesSerialExecution
= new CompletableFuture_Parallel();
completableFuturesSerialExecution.executeCompletableFuturesInParallel();
}
private void executeCompletableFuturesInParallel() throws InterruptedException, ExecutionException {
List<CompletableFuture<String>> completableFutures = new ArrayList<>();
TaskImpl task = new TaskImpl();
for (int i = 0; i < 5; i++) {
CompletableFuture<String> completableFuture = executeRule(i, task);
completableFutures.add(completableFuture);
}
CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]))
.thenAccept(__ -> System.out.println("All futures completed successfully"));
}
private CompletableFuture<String> executeRule(int i, TaskImpl customRuleNumber)
throws InterruptedException, ExecutionException {
return customRuleNumber.executeTask(i);
}
abstract class Task {
abstract CompletionStage<String> executeTask(int n);
}
class TaskImpl extends Task {
CompletableFuture<String> executeTask(int n) {
System.out.println("Executing Task number = " + n);
return CompletableFuture.completedFuture("" + n);
};
}
}
Output:
Executing Task number = 0
Executing Task number = 1
Executing Task number = 2
Executing Task number = 3
Executing Task number = 4
All futures completed successfully
Executing Completable Futures in Parallel : return after any of the futures complete
package completables;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
public class CompletableFuture_Parallel_anyOf {
public static void main(String[] args) throws InterruptedException, ExecutionException {
CompletableFuture_Parallel_anyOf completableFuturesSerialExecution
= new CompletableFuture_Parallel_anyOf();
completableFuturesSerialExecution.executeCompletableFuturesInParallel();
}
private void executeCompletableFuturesInParallel() throws InterruptedException, ExecutionException {
List<CompletableFuture<String>> completableFutures = new ArrayList<>();
TaskImpl task = new TaskImpl();
for (int i = 0; i < 5; i++) {
CompletableFuture<String> completableFuture = executeRule(i, task);
completableFutures.add(completableFuture);
}
CompletableFuture.anyOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]))
.thenAccept(__ -> System.out.println("Future(Optionally others also) completed successfully"));
}
private CompletableFuture<String> executeRule(int i, TaskImpl customRuleNumber)
throws InterruptedException, ExecutionException {
return customRuleNumber.executeTask(i);
}
abstract class Task {
abstract CompletionStage<String> executeTask(int n);
}
class TaskImpl extends Task {
CompletableFuture<String> executeTask(int n) {
System.out.println("Executing Task number = " + n);
return CompletableFuture.completedFuture("" + n);
};
}
}
Output:
Executing Task number = 0
Executing Task number = 1
Executing Task number = 2
Executing Task number = 3
Executing Task number = 4
Future(Optionally others also) completed successfully