Parallel Streams

A serial stream processes elements in an ordered manner, one at a time.

A parallel stream uses multiple threads to process data concurrently.

By default, the no of available CPUs affects the no of threads available for parallel stream.

Creating Parallel Streams

Using parallel() to create parallel stream from existing stream :

Stream<String> stream = Stream.of("Germany", "England", "China","Denmark", "Brazil");
Stream<String> parallelStream = stream.parallel();

Using parallelStream() to create a parallel stream from a collection :

Stream<String> parallelStream2 = Arrays.asList("Germany", "England",
        "China", "Denmark", "Brazil").parallelStream();

Examples of using Parallel Streams

Example 1 : Sum of numbers using parallelStream()

int sum = Arrays.asList(1,2,3,4,5).parallelStream().mapToInt(n -> n).sum();
System.out.println(sum);

Output :

15

Example 2 : Find count of Prime numbers using parallel()

import java.util.stream.IntStream;

public class PrimeNumberDemo {

  public static void main(String[] args) {
    final long primeCount = IntStream.range(1, 100)
        .parallel()
        .filter(number -> isPrime(number)).count();
    System.out.println("Count of prime numbers = " + primeCount);
  }

  public static boolean isPrime(final int number) {
    return number > 1
        && IntStream.rangeClosed(2, (int) Math.sqrt(number)).noneMatch(
            divisor -> number % divisor == 0);
  }
}

Output :

Count of prime numbers = 25

Order difference between Serial vs Parallel processing

A serial stream processes elements in a serial manner as shown in below example :

Arrays.asList("Germany", "England",
    "China", "Denmark", "Brazil").stream().forEach(System.out::println);

Output :

Germany
England
China
Denmark
Brazil
 
However, with a ParallelStream the data is processed in multiple threads. So, the order of the elements may vary as shown in following example:

Arrays.asList("Germany", "England",
    "China", "Denmark", "Brazil").parallelStream().forEach(System.out::println);

One of possible outputs :

China
Brazil
Denmark
England
Germany
 

Keeping order with ParallelStream

We can process elements in an ordered manner using forEachOrdered() method at the cost of performance.

Arrays.asList("Germany", "England",
    "China", "Denmark", "Brazil").parallelStream().forEachOrdered(System.out::println);

Output :

Germany
England
China
Denmark
Brazil
 
Although, there is performance cost associated with forEachOrdered(), this can be used with elements that need to be processed in parallel. In such case, stream operations before/after the forEachOrdered() can still use the performance improvements.
 

Performance improvement with Parallel Stream

Parallel stream provides performance improvements depending on the number of CPUs available.

For smaller streams, the improvement may be limited because of overhead associated with multitasking. In fact, a stream that processes a small amount of data may actually run slower with parallelStream().

Parallel streams are useful on independent stream operations.i.e, result of the operation on one element of the stream is not dependent on result of another element of the stream.

package com.topjavatutorial.java8examples;

import java.util.ArrayList;
import java.util.List;

public class ParallelStreamDemo {

  public static void main(String[] args) {
    List<Integer> numbers = new ArrayList<Integer>();
    for (int i = 0; i < 1000; i++) {
      numbers.add(i);
    }

    // Process data sequentially
    long startTime = System.currentTimeMillis();
    numbers.stream().forEach(i -> processElement(i));
    long endTime = System.currentTimeMillis();
    double timeTaken = (endTime - startTime) / 1000;
    System.out.println("Time taken with stream() : " + timeTaken
        + " milliseconds");

    // Process data in parallel
    startTime = System.currentTimeMillis();
    numbers.parallelStream().forEach(i -> processElement(i));
    endTime = System.currentTimeMillis();
    timeTaken = (endTime - startTime) / 1000;
    System.out.println("Time taken with parallelStream() : " + timeTaken
        + " milliseconds");
  }

  private static void processElement(int num) {
    try {
      Thread.sleep(10);
    } catch (InterruptedException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
  }

}

Output :

Time taken with stream() : 12.0 milliseconds
Time taken with parallelStream() : 2.0 milliseconds
 

© 2016, https:. All rights reserved. On republishing this post, you must provide link to original post