Parallel stream pitfalls and how to avoid them

Since streams were introduced in Java 8, we now have a clean and functional way of processing data. It gets even better when the processing task can be done in parallel with as much as adding .parallel() right after creating the stream.

The underlying risk lies in the implementation of parallel streams and the threads they use. By default, the parent thread is used as well as ForkJoin common pool.

Therefore, there is a chance of a deadlock or some starvation if there are any compute intensive or blocking parts to the stream processing.

We can check it by running the following code:

int result = IntStream.range(0, 1_000_000_000)
        .boxed()
        .parallel()
        .peek(val -> System.out.println(Thread.currentThread().getName()))
        .reduce((x,y)->x+2*y)
        .orElse(0);

The computation itself is not significant, but by printing the names of the processing threads, we can verify that ForkJoinPool.commonPool-worker threads are used (as well as the main thread).

If we use another parallel stream in another, unrelated part of our codebase, it will compete with the computation above for the aforementioned threads.

There is no API to provide a custom thread pool to the parallel stream, but there is a way by wrapping it in custom ForkJoinPool.

Lets take the computation above and make a callable out of it:

Callable<Integer> task = () -> IntStream.range(0, 1_000_000_000)
        .boxed()
        .parallel()
        .map(x -> x * 5)
        .peek(val -> System.out.println(Thread.currentThread().getName()))
        .reduce((x, y) -> x + 2 * y)
        .orElse(0);

Now we can create a custom pool and use it to run the task above:

ForkJoinPool pool = new ForkJoinPool(POOL_SIZE);
int result = pool.submit(task).get();

Running the code above with pool size of 8, we can now verify that threads ForkJoinPool-1-worker-0 to ForkJoinPool-1-worker-7 are used.

This way we essentially force the parallel stream to have a ForkJoin as a parent thread, instead of a regular thread, so ForkJoinPool.commonPool is not used. Hooray!