From Future to CompletableFuture
- Threads are the basic unit for doing work in the JVM
- Creating threads is expensive, so a long time ago, thread pools were created.
- More threads aren’t always better
- If there were no blocking I/O, then the ideal world would be #threads == #CPU (ie. logical cores that takes into consideration of hyper-threading). This is because switching the thread being executed on a CPU has a cost (saving and loading state from memory, etc). The operating system and JVM have to do work to enable this context switch.
- Java 7 introduced the ForkJoinPool implementation of ExecutorService, which understands this cost and always has #threads == #CPUs. As tasks are submitted, it runs newer tasks first (LIFO) to minimize context switching (since newer tasks are more likely to refer to data that is still in L1/L2 cache). It also allows threads to steal work from other threads’ queues in a FIFO manner to prevent old tasks from starving.
- Future class presents a clean API for the thread executing a Runnable and the thread waiting for the result to communicate, so you don’t have to deal with locks, volatile variables, notify(), etc. (i.e. the super low level concurrency primitives that Java provides). Sadly, the Future class does not provide an easy way to transform or compose Future
instances. For example: To transform Future to Future , you need to first call the get() method and it will block your thread until the result returned.
|
1 2 |
Future<User> u = getUser(int id); Boolean hasAddress = u.get().hasAddress(); |
The Java 8 CompletableFuture takes Java multithreading to new heights. For the example above, now you can rewrite it like below. The CompletableFuture.thenApply() method allows you to transform a Future<T> to a Future<U> without blocking. It also has methods to compose CompletableFutures together, like thenCombine. CompletableFuture also also plays very nicely with both lambdas and functional interfaces in general. With CompletableFuture, you can turn your IO bound method calls into non-blocking calls. So, your caller thread will not been blocked in waiting.
|
1 2 |
CompletableFuture<User> u = getUser(int id); CompletableFuture<Boolean> hasAddress = u.thenApply(User::hasAddress); |
Combine 2 CompletableFutures
If you have more than 2 remote calls and both are needed to complete to continue on the next method. You can use thenCombine() method. See example below. The creatDisplayVos method will be invoked when both userFutures and coverageFutures are completed.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
CompletableFuture<List<User>> userFutures = UserService.getAllUsers(); CompletableFuture<List<Coverage>> coverageFutures = CoverageService.getAllCoverages(); // Nothing blocks in the next line CompletableFuture<List<DisplayVo>> displayVos = users.thenCombine(coverages, this::createDisplayVos); List<DisplayVo> createDisplayVos(List<User> users, List<Coverage> coverages) { assert users.size() == coverages.size(); List<DisplayVo> results = new ArrayList<DisplayVo>(); for(int i=0; i < users.size(); i++){ User user = users.get(i); Coverage coverage = coverages.get(i); results.add(new DisplayVo(user, coverage)); } return results; } |
The problem of the above code is that we need to wait til both userFutures and coverageFutures completed before invoke the createDisplayVos method. And the createDisplayVos method uses ith element of users and coverages to create the ith DisplayVo. So, we can improve the code above further without introducing a waiting point. See the improvement below:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
List<CompletableFuture<Coverage> coverages = CoverageService.getAllCoverages(); // fetch user one after one instead of all together. List<CompletableFuture<User>> users = userIds.stream().map(userId -> userService.getUser(userId)).collect(Collectors.toList()); int size = Math.min(coverages.size(), users.size()); Stream<CompletableFuture<DisplayVo>> result = IntStream.range( 0, size) .mapToObj( (int i) -> { CompletableFuture<User> user = users.get(i); CompletableFuture<Coverage> coverage = coverages.get(i); return user.thenCombine(coverage, this::createDisplayVo); }); // Only change here is to accept single items, not Lists public DisplayVo createDisplayVo(User user, Coverage coverage) { return new DisplayVo(user, coverage); } |
Timeout
|
1 2 3 4 5 6 7 8 9 |
CompletableFuture futureCount = CompletableFuture.supplyAsync( () -> { try { // Simulate long running task Thread.sleep(5000); } catch (InterruptedException e) { } return 10; }); |
- CompletableFuture.supplyAsync allows you to run a task asynchronously on Java’s ForkJoinPool and also has the option to supply your own Executor if you want more control on the ThreadPool.
Connect with us