Force Java stream to execute part of a pipeline early to submit blocking tasks to a thread pool

Question

I have a list of objects that I want to process and Java8 stream API looks the most clean and readable way to do it.

But some operations I need to do on these objects include blocking IO (like reading a database) – so I’d like to submit those operations to a threadpool with a couple dozens threads.

At first I thought about doing something along the lines of:

myObjectList
    .stream()
    .filter(wrapPredicate(obj -> threadPoolExecutor.submit(
            () -> longQuery(obj)          // returns boolean
    ).get())                              // wait for future & unwrap boolean
    .map(filtered -> threadPoolExecutor.submit(
            () -> anotherQuery(filtered)  // returns Optional
    ))
    .map(wrapFunction(Future::get))
    .filter(Optional::isPresent)
    .map(Optional::get)
    .collect(toList());

The wrapPredicate and wrapFunction are just for checked exception rethrowing.

But, obviously, calls to Future.get() will block the stream’s thread until the query finishes for a given object and the stream won’t progress until then. So only one object is processed at a time and the threadpool doesn’t make sense.

I could use a parallel stream, but then I’d need to hope that the default ForkJoinPool is enough for this. Or just increase "java.util.concurrent.ForkJoinPool.common.parallelism", but I do not want to change the whole application’s settings for the sake of that stream. I could create the stream in a custom ForkJoinPool, but I see it does not guarantee that level of parallelism.

So I ended up with something like that, just to guarantee that all needed tasks are submitted to the threadPool before waiting for futures to be finished:

myObjectList
    .stream()
    .map(obj -> Pair.of(obj, threadPoolExecutor.submit(
                    () -> longQuery(obj)             // returns boolean
        ))
    )
    .collect(toList()).stream()                      // terminate stream to actually submit tasks to the pool
    .filter(wrapPredicate(p -> p.getRight().get()))  // wait & unwrap future after all tasks are submitted
    .map(Pair::getLeft)
    .map(filtered -> threadPoolExecutor.submit(
            () -> anotherQuery(filtered)             // returns Optional
    ))
    .collect(toList()).stream()                      // terminate stream to actually submit tasks to the pool
    .map(wrapFunction(Future::get))                  // wait & unwrap futures after all submitted
    .filter(Optional::isPresent)
    .map(Optional::get)
    .collect(toList());

Is there any obviously better way to achieve this?

A more elegant way to tell the stream “execute the pipelined steps until now for every object in the stream now” and then keep processing other than .collect(toList()).stream() and a better way to filter on the effect of a Future than packing it into Apache Commons Pair to filter on Pair::getRight later? Or perhaps a totally different approach to the problem?

Answer 1

You can greatly simplify your code by using

myObjectList.stream()
    .map(obj -> threadPoolExecutor.submit(
                    () -> longQuery(obj)? anotherQuery(obj).orElse(null): null))
    .collect(toList()).stream()
    .map(wrapFunction(Future::get))
    .filter(Objects::nonNull)
    .collect(toList());

One point is that there will be no improvement in concurrency, if you submit anotherQuery at a later time to the same executor. Therefore, you can execute it directly after longQuery returned true. At this point, obj is still in scope so you may use it for anotherQuery.

By extracting the result of the Optional, using null as representation of a failure, we can get the same representation of absent results, be it because longQuery returned false or anotherQuery returned an empty Optional. So all we have to do after extracting the Future’s result, is to .filter(Objects::nonNull).

The logic that you have to submit the jobs first, collecting the Futures, before getting the actual results does not change. There is no way around it anyway. All that other convenience methods or frameworks can offer, is to hide the actual temporary storage of these objects.

Answer 2

I think the main question's answer is no. In order to "execute" a stream, you need a terminal operation. But there might be room for improvement.

You can at least get rid of the pair by collecting to a map instead of a list:

stream.collect(toMap(Function.identity(),
                     obj -> threadPoolExecutor.submit(() -> longQuery(obj))))
      .entrySet()
      .stream()
      .filter(wrapPredicate(entry -> entry.getValue().get()))
      .map(Entry::getKey)
      ...

Note that this only works if none of the processed objects are equal to another. It makes the code slightly shorter and easier to read as you don't have to create the pair/entry yourself.

Answer 3

You can specify thread-pool for Java 8 parallel streams. You dont have to change application settings. More info: https://stackoverflow.com/a/22269778/7123191.

Answer 4

You can at least get rid of the pair by collecting to a map instead of a list:

stream.collect(toMap(Function.identity(),
                     obj -> threadPoolExecutor.submit(() -> longQuery(obj))))
      .entrySet()
      .stream()
      .filter(wrapPredicate(entry -> entry.getValue().get()))
      .map(Entry::getKey)
      ...