Java stream with parallel elements and sequential elements

Question

I want to run a Java stream in which the first element filters in parallel to the second and third element. The second and third element can not filter parallel, because the use a shared ressource.

int numberOfFilteredElements = Stream.of(firstElement, secondElememt, thirdElement)
  .filter(element -> element.someFilter())
  .collect(Collectors.toSet())
  .size();

A Java stream can be parallel or sequential. Concating streams are only parallel if all streams are parallel. So I can not create create a parallel and sequential stream and concat them.

Any other idea how I can solve this with streams? Or do I need to need use locks like ReentrantLock in someFilter() and run the stream parallel for all elements?

This looks like a problem that would be much easier to solve without streams. — Louis Wasserman
– Louis Wasserman, Commented Mar 31, 2021 at 18:34
I thought the same but was interessted if there would be a nice solution with streams I could learn. — Martin
– Martin, Commented Mar 31, 2021 at 18:44
Not with streams. Method someFilter could - of course - synchronize or lock, but I am not sure how the stream API would react to this. — Turing85
– Turing85, Commented Mar 31, 2021 at 18:52
If it is only about filter then there is one way to achieve it. You can check the hashcode of element and decide to enter in a synchronized block or not. — Aniket Sahrawat
– Aniket Sahrawat, Commented Mar 31, 2021 at 18:55

Most Noble Rabbit · Accepted Answer · 2021-04-01 09:57:20Z

1

We can say:

First element is Stream #1.
Second and third elements are Stream #2.

Then we can create a Parent Parallel Stream that consists of Stream #1, Stream #2:

    Predicate<Integer> someFilter = i -> {
        try { Thread.sleep(4000); } catch (InterruptedException e) {e.printStackTrace();}
        logger.info(i + " filtered on thread id: " + Thread.currentThread().getId());
        return i != 7;
    };
    Stream<Integer> stream1 = IntStream.rangeClosed(0, 3).boxed().filter(someFilter);
    Stream<Integer> stream2 = IntStream.rangeClosed(-3, -1).boxed().filter(someFilter);

    int numberOfFilteredElements = Stream.of(stream1, stream2)
            .parallel()
            .flatMap(Function.identity())
            .peek(integer -> logger.info("Peeked: " + integer.toString()))
            .collect(Collectors.toSet())
            .size();
    System.out.println("Set size is: " + numberOfFilteredElements);

Output:

2021-04-01 12:56:34.084  INFO 26336 --- [           main] com.example.demo.DemoApplication         : Started DemoApplication in 2.966 seconds (JVM running for 5.717)
2021-04-01 12:56:38.098  INFO 26336 --- [           main] com.example.demo.DemoApplication         : -3 filtered on thread id: 1
2021-04-01 12:56:38.098  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : 0 filtered on thread id: 22
2021-04-01 12:56:38.100  INFO 26336 --- [           main] com.example.demo.DemoApplication         : Peeked: -3
2021-04-01 12:56:38.100  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : Peeked: 0
2021-04-01 12:56:42.100  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : 1 filtered on thread id: 22
2021-04-01 12:56:42.100  INFO 26336 --- [           main] com.example.demo.DemoApplication         : -2 filtered on thread id: 1
2021-04-01 12:56:42.100  INFO 26336 --- [           main] com.example.demo.DemoApplication         : Peeked: -2
2021-04-01 12:56:42.100  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : Peeked: 1
2021-04-01 12:56:46.101  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : 2 filtered on thread id: 22
2021-04-01 12:56:46.101  INFO 26336 --- [           main] com.example.demo.DemoApplication         : -1 filtered on thread id: 1
2021-04-01 12:56:46.101  INFO 26336 --- [           main] com.example.demo.DemoApplication         : Peeked: -1
2021-04-01 12:56:46.101  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : Peeked: 2
2021-04-01 12:56:50.102  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : 3 filtered on thread id: 22
2021-04-01 12:56:50.102  INFO 26336 --- [onPool-worker-1] com.example.demo.DemoApplication         : Peeked: 3
Set size is: 7

Notice that Stream #1 (which can be your first element) runs in parallel to Stream #2 (which can be second and third elements).

edited Apr 1, 2021 at 9:57

answered Apr 1, 2021 at 9:40

Most Noble Rabbit

2,7862 gold badges8 silver badges12 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

Holger Over a year ago

It is very dangerous to assume that a particular limitation (the current implementation’s inability to parallelize sub-streams in flatMap) will persist all the time.

Most Noble Rabbit Over a year ago

Thanks for the comment. I am confused though. I don't see why would we expect flatMap to parallelize the sub-streams. I thought it specifies to parallelize just the parent stream, not its elements, no?

Holger Over a year ago

No, it does not specify anywhere that sub-streams are processed sequentially. That’s just a detail of the current implementation.

Collectives™ on Stack Overflow

Java stream with parallel elements and sequential elements

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related