Spring boot with Async API call and time comparison
This blog demonstrate the use of java CompletableFuture and Async annotation with spring boot.
Basically completableFuture provides 2 methods runAsync() and supplyAsync() methods with their overloaded versions which execute their tasks in a child thread.
CompletableFuture without any custom exectuor executes the tasks in a separate thread obtained from the ForkJoinPool.commonPool().
We can also create a custom thread pool and pass it to runAsync() and supplyAsync() methods to let them execute their tasks in a thread obtained from the custom thread pool .
All the methods in the CompletableFuture class has two overloaded version One with executor and one without it.
For example:
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)static CompletableFuture<Void> runAsync(Runnable runnable)static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn);
- Spring Async annotation with compeltable future using common pool
Controller Endpoint:
@GetMapping("asyncwithcompeltablefutureandcommonpool")
private Result getAsyncWithCompeltableFutureAndCommonPool() throws InterruptedException {AsyncRequestContext asyncRequestContext = new AsyncRequestContext();// multiple asynchronous lookups
long start = System.currentTimeMillis();CompletableFuture<ToDo> result1 = service.findToDo();CompletableFuture<List<Map<String, Object>>> result2 = service.getUserComments();CompletableFuture.allOf(result1, result2).thenApply(r -> {
Result result = new Result();
List<Map<String, Object>> secondResult = new ArrayList<>();
ToDo firstResult = null;
try {
secondResult = result2.get();
firstResult = result1.get();
}
long end = System.currentTimeMillis();
LOG.info("asyncwithcompeltablefutureandcommonpool future with common pool took " + (end - start) + "ms");
result = service.marshallResponse(firstResult, secondResult);
asyncRequestContext.setResult(result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
return null;
}).join();
return asyncRequestContext.getResult();
}
Service :
private static final Logger logger = LoggerFactory.getLogger(AsyncDemoService.class);
private final RestTemplate restTemplate;
public AsyncDemoService(RestTemplateBuilder restTemplateBuilder) {
this.restTemplate = restTemplateBuilder.build();
}@Async
public CompletableFuture<ToDo> findToDo() throws InterruptedException {
logger.info("findToDo thread " + Thread.currentThread().getName());
String url = "https://jsonplaceholder.typicode.com/todos/1"; ToDo result = restTemplate.getForObject(url, ToDo.class); return CompletableFuture.completedFuture(result);
}@Async
public CompletableFuture<List<Map<String, Object>>> getUserComments() throws InterruptedException {
logger.info("getUserComments thread " + Thread.currentThread().getName());
String url = "https://jsonplaceholder.typicode.com/comments?postId=1";
List<Map<String, Object>> result = restTemplate.getForObject(url, List.class); return CompletableFuture.completedFuture(result);
}
Output
If you see below output, there are two child thread started AsyncDemo-1 and AsyncDemo-2 (I have added Prefix while using declaring the Executor in spring boot main class) and time taken is 1818ms
2. Async with compeltablefuture and custom executor thread without Spring @Async Annotation
Controller:
@GetMapping("asyncwithcompeltablefutureandcustompool")
private Result getAsyncWithCompeltableFutureAndCustomPool() throws Exception {ExecutorService executor = Executors.newFixedThreadPool(10); AsyncRequestContext asyncRequestContext = new AsyncRequestContext();// multiple asynchronous lookups
long start = System.currentTimeMillis();CompletableFuture<ToDo> result1 = service.findToDoWithoutAsyncAnnotation(executor);CompletableFuture<List<Map<String, Object>>> result2 = service.getUserCommentsWithoutAsyncAnnotation(executor);CompletableFuture.allOf(result1, result2).thenAccept(r -> {
Result result = new Result();
try {
List<Map<String, Object>> secondResult = result2.get();
ToDo firstResult = result1.get();
long end = System.currentTimeMillis();
LOG.info("asyncwithcompeltablefutureandcustompool future with custom pool took " + (end - start) + "ms");
result = service.marshallResponse(firstResult, secondResult);
asyncRequestContext.setResult(result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}).join();
return asyncRequestContext.getResult();
}
Service:
public CompletableFuture<ToDo> findToDoWithoutAsyncAnnotation(ExecutorService executor)
throws InterruptedException {CompletableFuture<ToDo> future = CompletableFuture.supplyAsync(new Supplier<ToDo>() {@Override
public ToDo get() {
logger.info("findToDo thread " + Thread.currentThread().getName()); String url = "https://jsonplaceholder.typicode.com/todos/1"; ToDo result = restTemplate.getForObject(url, ToDo.class); return result;
}
}, executor);
return future;
}public CompletableFuture<List<Map<String, Object>>>
getUserCommentsWithoutAsyncAnnotation(ExecutorService executor)
throws InterruptedException {CompletableFuture<List<Map<String, Object>>> future = CompletableFuture.supplyAsync(new Supplier<List<Map<String, Object>>>() {@Override
public List<Map<String, Object>> get() {
logger.info("getUserComments thread " + Thread.currentThread().getName()); String url = "https://jsonplaceholder.typicode.com/comments?postId=1"; List<Map<String, Object>> result = restTemplate.getForObject(url, List.class);return result;
}
}, executor);
return future;
}
If you see in the above Controller and service code we have defined our own executor wit fixed thread pool of 10 and passing it the the CompletableFuture to let them use our own thread instead from the common pool
Output:
If you see below output, there are two child thread started thread-pool-1-thread-1 and thread-pool-1-thread-2 and time taken is 153ms which is less than the first approach we have used.
3. Async with compeltablefuture and common thread pool without Spring @Async Annotation
Controller:
@GetMapping("asyncwithcompeltablefutureandcommonpoolwithoutasyncannotation")
private Result getAsyncWithCompeltableFutureAndCustomPool() throws Exception {AsyncRequestContext asyncRequestContext = new AsyncRequestContext();// multiple asynchronous lookups
long start = System.currentTimeMillis();CompletableFuture<ToDo> result1 = service.findToDoWithoutAsyncAnnotation();CompletableFuture<List<Map<String, Object>>> result2 = service.getUserCommentsWithoutAsyncAnnotation();CompletableFuture.allOf(result1, result2).thenAccept(r -> {
Result result = new Result();
try {
List<Map<String, Object>> secondResult = result2.get();
ToDo firstResult = result1.get();
long end = System.currentTimeMillis();
LOG.info("asyncwithcompeltablefutureandcustompool future with custom pool took " + (end - start) + "ms");
result = service.marshallResponse(firstResult, secondResult);
asyncRequestContext.setResult(result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}).join();
return asyncRequestContext.getResult();
}
Service:
public CompletableFuture<ToDo> findToDoWithoutAsyncAnnotation()
throws InterruptedException {CompletableFuture<ToDo> future = CompletableFuture.supplyAsync(new Supplier<ToDo>() {@Override
public ToDo get() {
logger.info("findToDo thread " + Thread.currentThread().getName());String url = "https://jsonplaceholder.typicode.com/todos/1";ToDo result = restTemplate.getForObject(url, ToDo.class);return result;
}
});
return future;
}public CompletableFuture<List<Map<String, Object>>>
getUserCommentsWithoutAsyncAnnotation()
throws InterruptedException {CompletableFuture<List<Map<String, Object>>> future = CompletableFuture.supplyAsync(new Supplier<List<Map<String, Object>>>() {@Override
public List<Map<String, Object>> get() {
logger.info("getUserComments thread " + Thread.currentThread().getName());String url = "https://jsonplaceholder.typicode.com/comments?postId=1";List<Map<String, Object>> result = restTemplate.getForObject(url, List.class);return result;
}
});
return future;
}
If you see in the above Controller and service code we have defined our own executor wit fixed thread pool of 10 and passing it the the CompletableFuture to let them use our own thread instead from the common pool
Output:
If you see below output, there are two child thread(from Common thread pool) started common-worker-1 and common-worker-2 and time taken is 726ms which is less than the first approach and more than second appraoch.
You can use timeout on each API call using below code:
CompletableFuture<ToDo> result1 = service.findToDoWithoutAsyncAnnotation();
ToDo firstResult = result1.get(1000, TimeUnit.MILLISECONDS);
With Java 9 two new methods have been provided to handle the timouts with CompletableFuture
a) CompletableFuture#orTimeout
It will throw java.util.concurrent.ExecutionException if the future doesn’t complete within a specified timeout.
public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit)Example:CompletableFuture<ToDo> future = CompletableFuture.supplyAsync(this::getToDo)
.orTimeout(2, TimeUnit.SECONDS);ToDo result = future.get();
b) CompletableFuture#completeOnTimeout
public CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit)Here, default value can be returned once the timeout is reached:
ToDo defaultToDo = new ToDo();CompletableFuture<ToDo> future = CompletableFuture.supplyAsync(this::getToDo)
.completeOnTimeout(defaultToDo, 2,TimeUnit.SECONDS);ToDo result = future.get();
Exception handling with completable future
Three different methods have been provided to handle exception with CompletableFuture
public CompletableFuture <T> exceptionally(Function <Throwable, ? extends T> function);
public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> bifunction);
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);exceptionally()
It returns a new CompletionStage that mean when the stage completes with exception it returns with the exception as the argument to the supplied function. Else, if the stage completes normally, then the returned stage also completes normally with the same value. So if there’s no exception then exceptionally( ) stage is skipped otherwise it will be executed.
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {return 10 / 0;}).exceptionally(exception -> {System.err.println("exception: " + exception);return 1;});
Output
exception: java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zerohandle()
The stage returned by the handle() method is always executed whether exception occurs or not. If stage completes without exception it returns with exception as null
CompletableFuture.supplyAsync(() -> {return 10 / 0;}).handle((input, exception) -> {if (exception != null) {System.out.println(exception);return 1;}return 0;}).thenApply(input -> input + 1).thenAccept(System.out::println);
Output:
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero2
whenComplete()
whenComplete() method accept BiConsumer, whereas handle() methods accept BiFunction which mean handle method return the result in case of exception but whenComplete method cannot return result. If any exception comes from the previous stage then it will be passed through to the next pipeline as it is.
CompletableFuture.supplyAsync(() -> {return 10 / 0;}).whenComplete((input, exception) -> {if (exception != null) {System.out.println(exception);}}).thenApply(input -> input + 1).thenAccept(System.out::println);
Output:
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zeroBelow is the link to github