Calling two functions in parallel in Java 8

user_1823822 0
Viewed 1482

I have a use case in my Spring boot application as follows:

I would like to fetch the id field value from the response with the following function:

String id = getIdFromResponse(response);

If I don't get any id in the response, then I check if the id field is present in the request argument with the following function:

String id = getIdFromRequest(request);

As of now, I am invoking them sequentially. But I would like to make these two functions run parallelly, I would like to stop as soon as I get an id from either of them.

I am wondering if there is any way to implement this using streams in Java 8.

java-8 spring-boot multithreading spring java
3 Answers

You can use something like this:

String id = Stream.<Supplier<String>>of(
        () -> getIdFromResponse(response), 
        () -> getIdFromRequest(request)
    )
    .parallel()
    .map(Supplier::get)
    .filter(Objects::nonNull)
    .findFirst()
    .orElseThrow():

The suppliers are needed, because when you don't use them, then both requests are still executed sequentially.

I also assumed that your methods return null when nothing is found, so I had to filter these values out with .filter(Objects::nonNull).

Depending on your use case, you can replace .orElseThrow() with something different, like .orElse(null)

user_55150600

There is no need to use Stream API as long as there exists a method exactly for this.

ExecutorService::invokeAny(Collection<? extends Callable<T>>)

Executes the given tasks, returning the result of one that has completed successfully (i.e., without throwing an exception), if any do. Upon normal or exceptional return, tasks that have not completed are cancelled.

List<Callable<String>> collection = Arrays.asList(
    () -> getIdFromResponse(response),
    () -> getIdFromRequest(request)
);

// you want the same number of threads as the size of the collection
ExecutorService executorService = Executors.newFixedThreadPool(collection.size());
String id = executorService.invokeAny(collection);

Three notes:

  • There is also an overloaded method with timeout throwing TimeoutException if no result is available in time: invokeAny(Collection<? extends Callable<T>>, long, TimeUnit)
  • You need to handle ExecutionException and InterruptedException from the invokeAny method.
  • Don't forget to close the service once you are done
user_37649650

If you want to be in full control over when to enable the alternative evaluation, you may use CompletableFuture:

CompletableFuture<String> job
    = CompletableFuture.supplyAsync(() -> getIdFromResponse(response));
String id;
try {
    id = job.get(300, TimeUnit.MILLISECONDS);
}
catch(TimeoutException ex) {
    // did not respond within the specified time, set up alternative
    id = job.applyToEither(
        CompletableFuture.supplyAsync(() -> getIdFromRequest(request)), s -> s).join();
}
catch(InterruptedException|ExecutionException ex) {
    // handle error
}

The second job is only submitted when the first did not complete within the specified time. Then, whichever job responds first will provide the result value.

user_27114880
Related