Interface Future
A Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation. The result can only be retrieved using method get when the computation has completed, blocking if necessary until it is ready. Cancellation is performed by the cancel method. Additional methods are provided to determine if the task completed normally or was cancelled. Once a computation has completed, the computation cannot be cancelled. If you would like to use a Future for the sake of cancellability but not provide a usable result, you can declare types of the form Future and return null as a result of the underlying task.
Sample Usage (Note that the following classes are all made-up.)
interface ArchiveSearcher < String search(String target); >class App < ExecutorService executor = . ; ArchiveSearcher searcher = . ; void showSearch(String target) throws InterruptedException < Callabletask = () -> searcher.search(target); Future future = executor.submit(task); displayOtherThings(); // do other things while searching try < displayText(future.get()); // use future >catch (ExecutionException ex) < cleanup(); return; >> > The FutureTask class is an implementation of Future that implements Runnable , and so may be executed by an Executor . For example, the above construction with submit could be replaced by:
FutureTask future = new FutureTask<>(task); executor.execute(future); Memory consistency effects: Actions taken by the asynchronous computation happen-before actions following the corresponding Future.get() in another thread.
Write Clean Asynchronous Code With CompletableFuture Java-8
Java 8 has introduced a lot of features. With Addition of CompletableFuture . Writing Clean & Readable Asynchronous code has become much more easier. CompletableFuture has more than 50 methods which makes it very useful.
The code for this post is available for download here.
CompletableFuture
CompletableFuture is an implementation of the Future & CompletionStage interface but with a lot of modern Features. It Supports lambdas and takes advantage of non-blocking methods via callbacks and promotes asynchronous reactive programming model. CompletableFuture allows us to write 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. CompletableFuture is inspired from ListenableFuture in Guava and Are similar to Promise in java scripts.
Why CompletableFuture instead Of Future?
Callable and Future were introduced in Java 5. Future is placeholders for a result that hasn’t happened yet.Future Can use a Runnable or Callable instance to complete the submitted task. There are two methods to get actual value from Future.
get() : When this method is called, thread will wait for result indefinitely.
get(long timeout, TimeUnit unit) : When this method is called, thread will wait for result only for specified time.
There are multiple problems with Future
Blocking — The get method is blocking and need to wait until the computation is done. Future does not have any method that can notify on completion and does not have capability to attach a callback function.
Chaining & Composition — Many times we want to chain multiple future to complete long computation. You need to merger results and send results to another task. It’s Hard to implement such chaining with future.
Exception Handling — Future does not provide any construct for Exception Handling.
All these issues are addressed by CompletableFuture.
lets try different methods provided by CompletableFuture
Create Simple Completeable Future
The simplest way is to create CompleteableFuture is CompleteableFuture.completedFuture method which returns an a new, finished CompleteableFuture . Creating already Completed CompleteableFuture becomes very useful in many cases.
@Test
public void completableFutureThenApplyAccept() {
CompletableFuture.supplyAsync(this::findAccountNumber)
.thenApply(this::calculateBalance)
.thenApply(this::notifyBalance)
.thenAccept((i)->notifyByEmail()).join();
}
@Test
public void completableFutureApplyAsync() {
ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
ScheduledExecutorService newSingleThreadScheduledExecutor = Executors.newSingleThreadScheduledExecutor();
CompletableFuture completableFuture =
CompletableFuture
.supplyAsync(this::findAccountNumber,newFixedThreadPool)//will run on thread obtain from newFixedThreadPool
.thenApplyAsync(this::calculateBalance,newSingleThreadScheduledExecutor) //will run on thread obtain from newSingleThreadScheduledExecutor
.thenApplyAsync(this::notifyBalance);//will run on thread obtain from common pool
Integer balance = completableFuture.join();
assertEquals(Integer.valueOf(balance), Integer.valueOf(100));
}
CompletableFuture thenCompose and thenCombine
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public CompletableFuture findAccountNumber() {
sleep(1);
System.out.println(Thread.currentThread() + » findAccountNumber»);
return CompletableFuture.completedFuture(10);
}
public CompletableFuture calculateBalance(int accountNumber) {
System.out.println(Thread.currentThread() + » calculateBalance»);
sleep(1);
return CompletableFuture.completedFuture(accountNumber * accountNumber);
}
public CompletableFuture notifyBalance(Integer balance) {
System.out.println(Thread.currentThread() + «Sending Notification»);
sleep(1);
return CompletableFuture.completedFuture(balance);
}
private void sleep(int seconds) {
try {
TimeUnit.SECONDS.sleep(seconds);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Test
public void completableFutureThenCompose()
{
Integer join = findAccountNumber()
.thenComposeAsync(this::calculateBalance)
.thenCompose(this::notifyBalance)
.join();
assertEquals(new Integer(100), join);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public CompletableFuture findFirstName() {
return CompletableFuture.supplyAsync(() -> {
sleep(1);
return «Niraj»;
});
}
public CompletableFuture findLastName() {
return CompletableFuture.supplyAsync(() -> {
sleep(1);
return «Sonawane»;
});
}
@Test
public void completableFutureThenCombine() {
CompletableFuture thenCombine =
findFirstName().thenCombine(findLastName(), (firstName, lastname) -> {
return firstName + lastname;});
String fullName = thenCombine.join();
assertEquals(«NirajSonawane», fullName);
}
CompletableFuture allOf
In Many scenario we want to run run multiple task in parallel and want to do some processing after all of them are complete.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public CompletableFuture findSomeValue() {
return CompletableFuture.supplyAsync(() -> {
sleep(1);
return «Niraj»;
});
}
@Test
public void completableFutureAllof() {
List
IntStream.range(0, 5).forEach(num -> {
list.add(findSomeValue());
});
CompletableFuture allfuture = CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()]));//Created All of object
CompletableFuture allFutureList = allfuture.thenApply(val -> {
return list.stream().map(f -> f.join()).collect(Collectors.toList());
});
CompletableFuture futureHavingAllValues = allFutureList.thenApply(fn -> {
System.out.println(«I am here»);
return fn.stream().collect(Collectors.joining());});
String concatenateString = futureHavingAllValues.join();
assertEquals(«NirajNirajNirajNirajNiraj», concatenateString);
}
CompletableFuture Exception Handling
CompletableFuture thenApply = CompletableFuture
.supplyAsync(this::findAccountNumber)
.thenApply(this::calculateBalance)
.thenApply(this::notifyBalance)
Handel Exceptions using exceptionally
@Test
public void completableFutureExceptionally()
{
CompletableFuture thenApply = CompletableFuture.supplyAsync(this::findAccountNumber)
.thenApply(this::calculateBalance)
.thenApply(this::notifyBalance)
.exceptionally(ex -> {
System.out.println(«Got Some Exception «+ex.getMessage());
System.out.println(«Returning some default value»);
return 0;
});
Integer join = thenApply.join();
assertEquals(new Integer(0), join);
}
Handel Exceptions using Handel Method
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Test
public void completableFutureHandel()
{
CompletableFuture thenApply = CompletableFuture.supplyAsync(this::findAccountNumber)
.thenApply(this::calculateBalance)
.thenApply(this::notifyBalance)
.handle((ok, ex) -> {
System.out.println(«Code That we want to run in finally «);
if (ok != null) {
System.out.println(«No Exception !!»);
} else {
System.out.println(«Got Exception « + ex.getMessage());
return —1;
}
return ok;
});
}
@Test
public void completableFutureWhenComplete()
{
CompletableFuture.supplyAsync(this::findAccountNumber)
.thenApply(this::calculateBalance)
.thenApply(this::notifyBalance)
.whenComplete((i,t)->System.out.println(«finally action»));
}
TimeOut java 9 Improvement
While Working on Asynchronous Code, We Need to handel timeouts. We Can not wait forever to finish the task. Unfortunately we do not have anything in java 8 for timeouts.
Java 9 has added orTimeout and completeOnTimeout methods to handel this.
If the task does not complete in given time, a TimeoutException will be thrown.
@Test
public void completableFutureWhenComplete()
{
CompletableFuture.supplyAsync(this::findAccountNumber)
.orTimeout(1, TimeUnit.MINUTES);
}
The code for this post is available for download here.
Interface Future
A Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation. The result can only be retrieved using method get when the computation has completed, blocking if necessary until it is ready. Cancellation is performed by the cancel method. Additional methods are provided to determine if the task completed normally or was cancelled. Once a computation has completed, the computation cannot be cancelled. If you would like to use a Future for the sake of cancellability but not provide a usable result, you can declare types of the form Future and return null as a result of the underlying task.
Sample Usage (Note that the following classes are all made-up.)
interface ArchiveSearcher < String search(String target); >class App < ExecutorService executor = . ; ArchiveSearcher searcher = . ; void showSearch(String target) throws InterruptedException < Callabletask = () -> searcher.search(target); Future future = executor.submit(task); displayOtherThings(); // do other things while searching try < displayText(future.get()); // use future >catch (ExecutionException ex) < cleanup(); return; >> > The FutureTask class is an implementation of Future that implements Runnable , and so may be executed by an Executor . For example, the above construction with submit could be replaced by:
FutureTask future = new FutureTask<>(task); executor.execute(future); Memory consistency effects: Actions taken by the asynchronous computation happen-before actions following the corresponding Future.get() in another thread.