CODEWITHSUNDEEP
javaconcurrencyjava-8futurecompletable-future
syfro
syfro

Reputation: 121

Most efficient way to stream on list of Futures

I'm calling an async client method by streaming over a list of objects. The method returns Future.

What's the best way to iterate over the list of Futures returned after the call (so as to process those Future which comes first)?

Note: The async client only returns Future not CompletableFuture.

Following is the code:

List<Future<Object>> listOfFuture = objectsToProcess.parallelStream()
    .map((object) -> {
        /* calling an async client returning a Future<Object> */ })
    .collect(Collectors.toList());

Upvotes: 8

Views: 11123

Answers (2)

David Lilljegren
David Lilljegren

Reputation: 1939

I found the stream().parallel() tricky to control, depending on the data structure of origin it may not always run in parallel. Here's a way to fork join a stream of parallel tasks using CompletableFuture.join.

    var pool = new ForkJoinPool(4);
    IntFunction<CompletableFuture<IntStream>> makePair = i -> CompletableFuture.supplyAsync(() -> IntStream.of(i, i), pool);
    var res = IntStream.of(1, 2, 3, 4, 5).mapToObj(makePair).flatMapToInt(CompletableFuture::join).sum();
    System.out.println(res);

Upvotes: 0

Eugene
Eugene

Reputation: 120868

Having this list of List<Future<Object>>, I would submit it to a custom pool, instead of using the default stream parallel processing.

That is because the stream api uses a common pool for parallel processing and you will call get on those Futures(if it takes significant time for processing) - you will block all other stream operations that use parallel operations within your application until this one is done.

This would a bit like this:

forJoinPool.submit( () -> list.stream().parallel().map(future -> future.get()).collect(Collectors.toList())).get();

I would go with a custom pool like shown here

Upvotes: 7

Related Questions