4

This is a question I constantly ask myself when designing a data intensive application: When is it appropriate to use stream() over parallelStream()? Would it make sense to use both? How do I quantify the metrics and conditions to intelligently decide which ones to use at runtime. From what I understand, parallelStream() is a great facility to process entries in parallel but it all comes down to execution time and overhead. Does the end justify the means?

In my particular use case, do to the nature of the application, the velocity and volume of the data I am processing will be all over the place. There will be times where the volume is so large, my application would massively benefit from parallelizing the workload. Then there are times where a single thread will accomplish the task much more efficiently. I have profiled my application a dozen times and have had mixed results.

So this brings me to my question. Is there a way in Java 8 (or later) to switch between stream() and parallelStream() intelligently? I considered at one point defining boundaries on the data that would allow for alternating between the two but in the end, not every piece of equipment is designed the same. Some systems may deal with single threaded workload much better then others. And vice versa.

It might be relevant to mention that I am using Apache Kafka, using Kafka Streams with Spring Cloud Streams. For the most part, I feel like I have squeezed everything out of Kafka in terms of performance and want to focus internally on optimizing my own service.

1
  • Parallel streams can utilize multiple cores easily. If you don't have that, you don't benefit. Commented Feb 12, 2021 at 18:55

3 Answers 3

5

You can define a custom thread pool by implementing the (Executor) interface that increases or decreases the number of threads in the pool as needed. You can submit your parallelStream chain to it as shown here using a ForkJoinPool:

I've created a working example which prints the threads that are doing the work:

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;
import java.util.stream.LongStream;

public class TestParallel
{
  public static void main(String... args) throws InterruptedException, ExecutionException
  {
    testParallel();
  }
  
  
  static Long sum(long a, long b)
  {
    System.out.println(Thread.currentThread() + " - sum: " + a + " " + b);
    return a + b;
  }
  
  public static void testParallel() 
      throws InterruptedException, ExecutionException {
        
        long firstNum = 1;
        long lastNum = 10;

        List<Long> aList = LongStream.rangeClosed(firstNum, lastNum).boxed()
          .collect(Collectors.toList());

        System.out.println("custom: ");
        System.out.println();
        
        ForkJoinPool customThreadPool = new ForkJoinPool(4);
        long totalCustom = customThreadPool.submit(
          () -> aList.parallelStream().reduce(0L, TestParallel::sum)).get();
        
        System.out.println();
        System.out.println("standard: ");
        System.out.println();
        
        long totalStandard = aList.parallelStream().reduce(0L, TestParallel::sum);
        
        System.out.println();
        System.out.println(totalCustom + " " + totalStandard);
    }
}

Personally, if you want to get to that level of control, I'm not sure the streaming API is worth bothering with. It's not doing anything you can't do with Executors and concurrent libs. It's just a simplified facade to those features with limited capabilities.

Streams are kind of nice when you need to lay out a simple multi-step process in a little bit of code. But if all you are doing is using them to manage parallelism of tasks, the Executors and ExecutorService are more straightforward IMO. One thing I would avoid is pushing the number of threads above your machine's native thread count unless you have IO-bound processing. And if that's the case NIO is the more efficient solution.

What I'm not sure about is what the logic is that decides when to use multiple threads and when to use one. You'd have to better explain what factors come into play.

13
  • This is a good answer. IIRC, streams incurs quite a bit of overhead because it tries to clean up once the work is completed. Thank you for this suggestion. Commented Feb 4, 2020 at 22:52
  • I looked at the ForkJoinPool link. I don't understand why this approach for the parallel stream should run in parallel inside the pool instead of inside the global pool? Commented Feb 12, 2021 at 18:58
  • @ThorbjørnRavnAndersen Can you clarify what you mean by 'global thread pool'? This doesn't seem to be a standard term with regard to Java.
    – JimmyJames
    Commented Feb 12, 2021 at 19:09
  • @JimmyJames To my understanding the Stream parallelization uses a single, global data structure for handling the individual tasks, which cannot easily be overridden as you say it can. Commented Feb 14, 2021 at 9:20
  • @ThorbjørnRavnAndersen Can you please provide some technical reference to what you are referring to? Streams are a relatively new feature relative to threads and the concurrency packages.
    – JimmyJames
    Commented Feb 15, 2021 at 14:55
2

I don't know if this is useful but there is a design pattern called Bridge that decouples the abstraction from its implementation so you can, at runtime change between implementations.

A simple example would be a stack. For stacks where the total amount of data stored at one time is relatively small, it is more efficient to use an array. When the amount of data hits a certain point, it becomes better to use a linked-list. The stack implementation determines when it switches from one to the other.

For your case, it sounds like the processing would be behind some interface and based on the volume (do you know it before you start the processing?) your Processor class could use streams or parallel streams as appropriate.

2
  • A stack is great example. I would like to read more on that specification. The biggest question is when does it decide to alternate between those two implementations? Commented Feb 4, 2020 at 22:34
  • @user0000001 You make the decision because you're coding it. I hope I didn't give the impression that this is how all stacks are implemented; it's just an illustration of the pattern. So you need to decide when to choose the different implementation; maybe you already know how big your dataset is?
    – Matthew
    Commented Feb 5, 2020 at 13:34
0

StreamSupport.stream creates a new sequential or parallel Stream from a Spliterator (which in turn can be obtained from any Collection). Conditionally you can switch parallel processing on or off.

boolean parallel = true;
StreamSupport.stream(IntStream.range(0, 10).spliterator(), parallel).forEach(...);

Of course, you can also if-else if you have such a flag:

boolean parallel = findOutIfParallelizingIsWorthIt();
Stream<T> myStream = parallel ? myCollection.parallelStream() : myCollection.stream();
myStream.forEach(...);
2
  • 2
    I downvoted your answer, because it is really incomprehensive, adding very little value to the topic.
    – Andy
    Commented Feb 9, 2021 at 7:15
  • I up-voted it because it helped me.
    – fig
    Commented Apr 30, 2023 at 18:35

Not the answer you're looking for? Browse other questions tagged or ask your own question.