Using Parallel Streams - Java Stream API

Using Parallel Streams

Before exploring any more of the stream API, it will be helpful to discuss parallel streams. As has been pointed out previously in this book, the parallel execution of code via multicore processors can result in a substantial increase in performance. Because of this, parallel programming has become an important part of the modern programmer’s job. However, parallel programming can be complex and error-prone. One of the benefits that the stream library offers is the ability to easily—and reliably—parallel process certain operations.

Parallel processing of a stream is quite simple to request: just use a parallel stream. As mentioned earlier, one way to obtain a parallel stream is to use the parallelStream( ) method defined by Collection. Another way to obtain a parallel stream is to call the parallel( ) method on a sequential stream. The parallel( ) method is defined by BaseStream, as shown here:

S parallel()

It returns a parallel stream based on the sequential stream that invokes it. (If it is called on a stream that is already parallel, then the invoking stream is returned.) Understand, of course, that even with a parallel stream, parallelism will be achieved only if the environment supports it.

Once a parallel stream has been obtained, operations on the stream can occur in parallel, assuming that parallelism is supported by the environment. For example, the first reduce( ) operation in the preceding program can be parallelized by substituting parallelStream( ) for the call to stream( ):

Optional<Integer> productObj = myList.parallelStream().reduce((a,b) -> a*b);

The results will be the same, but the multiplications can occur in different threads.

As a general rule, any operation applied to a parallel stream must be stateless. It should also be non-interfering and associative. This ensures that the results obtained by executing operations on a parallel stream are the same as those obtained from executing the same operations on a sequential stream.

When using parallel streams, you might find the following version of reduce( ) especially helpful. It gives you a way to specify how partial results are combined:

<U> U reduce(U identityVal, BiFunction<U, ? super T, U> accumulator BinaryOperator<U> combiner)

In this version, combiner defines the function that combines two values that have been produced by the accumulator function. Assuming the preceding program, the following statement computes the product of the elements in myList by use of a parallel stream:

int parallelProduct = myList.parallelStream().reduce(1, (a,b) -> a*b, (a,b) -> a*b);

As you can see, in this example, both the accumulator and combiner perform the same function. However, there are cases in which the actions of the accumulator must differ from those of the combiner. For example, consider the following program. Here, myList contains a list of double values. It then uses the combiner version of reduce( ) to compute the product of the square roots of each element in the list.

// Demonstrate the use of a combiner with reduce()

import java.util.*;

import java.util.stream.*;

class StreamDemo3 {

public static void main(String[] args) {

// This is now a list of double values.

ArrayList<Double> myList = new ArrayList<>( );

myList.add(7.0);

myList.add(18.0);

myList.add(10.0);

myList.add(24.0);

myList.add(17.0);

myList.add(5.0);

double productOfSqrRoots = myList.parallelStream().reduce( 1.0,

(a,b) -> a * Math.sqrt(b), (a,b) -> a * b

);

System.out.println("Product of square roots: " + productOfSqrRoots);

}

}

Notice that the accumulator function multiplies the square roots of two elements, but the combiner multiplies the partial results. Thus, the two functions differ. Moreover, for this computation to work correctly, they must differ. For example, if you tried to obtain the product of the square roots of the elements by using the following statement, an error would result:

// This won't work.

double productOfSqrRoots2 = myList.parallelStream().reduce( 1.0,

(a,b) -> a * Math.sqrt(b));

In this version of reduce( ), the accumulator and the combiner function are one and the same. This results in an error because when two partial results are combined, their square roots are multiplied together rather than the partial results, themselves.

As a point of interest, if the stream in the preceding call to reduce( ) had been changed to a sequential stream, then the operation would yield the correct answer because there would have been no need to combine two partial results. The problem occurs when a parallel stream is used.

You can switch a parallel stream to sequential by calling the sequential( ) method, which is specified by BaseStream. It is shown here:

S sequential( )

In general, a stream can be switched between parallel and sequential on an as-needed basis. There is one other aspect of a stream to keep in mind when using parallel execution:

the order of the elements. Streams can be either ordered or unordered. In general, if the data source is ordered, then the stream will also be ordered. However, when using a parallel stream, a performance boost can sometimes be obtained by allowing a stream to be unordered. When a parallel stream is unordered, each partition of the stream can be operated on independently, without having to coordinate with the others. In cases in which the order of the operations does not matter, it is possible to specify unordered behavior by calling the unordered( ) method, shown here:

S unordered( )

One other point: the forEach( ) method may not preserve the ordering of a parallel stream. If you want to perform an operation on each element in a parallel stream while preserving the order, consider using forEachOrdered( ). It is used just like forEach( ).