Rules while working with stream in Java 8

First, let’s have a basic understanding of stream. Then we will have a look at the side effects that can occur while working with streams.

Streams represent a sequence of objects from a source, which supports aggregate operations. One thing to be notified while working with streams  is that, aggregate operation (intermediate operations) are lazy evaluated i.e. they do not start processing the content of the streams until the terminal operation commences. This enables the Java compiler and runtime to optimise how they process streams.

With Java 8, Collection interface has two methods to generate a Streams.

1. stream() − Returns a sequential stream considering collection as its source.

Given below is a very basic example in which array elements of type double which are less than 5 are converted into integer. The order in which the output received is fixed since the we are working with streams.

/**
* Basic example of streams
*/
Double[] doubleArray = {1.8, 2.0, 3.2, 4.5, 5.6, 6.0, 7.0, 8.9};
List<Double> listOfDouble =
new ArrayList<>(Arrays.asList(doubleArray));

listOfDouble
.stream()
.filter(element -> element < 5) .mapToInt(value -> value.intValue()).
forEach(System.out::println);

2. parallelStream() − Returns a parallel Streams considering collection as its source.

Parallel computing involves dividing a problem into subproblems, solving those problems simultaneously (in parallel, with each subproblem running in a separate thread), and then combining the results of the solutions to the subproblems. When a stream executes in parallel, the Java runtime partitions the streams into multiple substreams. Aggregate operations iterate over and process these substreams in parallel and then combine the results.

Now, in the example below, we have used parallel streams instead of streams. The order in which the elements are evaluated can vary each time it is run.

/**
* Basic example of parallel streams
*/
Double[] doubleArray = {1.8, 2.0, 3.2, 4.5, 5.6, 6.0, 7.0, 8.9};
List<Double> listOfDouble =
new ArrayList<>(Arrays.asList(doubleArray));

listOfDouble
.parallelStream()
.filter(element -> element < 5) .mapToInt(value -> value.intValue()).
forEach(System.out::println);

Side Effects :

1. Interference

Talking about side effects, Lambda expressions in stream operations should not interfere. Interference occurs when the source of a stream is modified while a pipeline processes the streams.

/**
* Example of interference
* This will through ConcurrentModificationException
*/
try {
List<String> listOfStrings =
new ArrayList<>(Arrays.asList("one", "two", "four"));

listOfStrings
.parallelStream()
.map(s -> {
listOfStrings.add("three");
return listOfStrings;
})
.forEach(System.out::println);
} catch (Exception e) {
System.out.println("Exception caught: " + e.toString());
}

The above example results in ConcurrentModificationException.  The same will occur even if we will use streams instead of parallel streams. The reason for such type of behaviour is lazy evaluation. This means that the pipeline in this example begins execution when the operation get is invoked, and ends execution when the get operation completes. When an attempt is made to modify the streams source during the execution of the pipeline, it throws runtime exception.

2. Stateful Lambda Expressions

A stateful lambda expression is one whose result depends on any state that might change during the execution of a pipeline. Understanding of the example below will give a clear understanding of this definition.

/**
* Example for stateful lambda expression
*/
Integer[] intArray = {1, 2, 3, 4, 5, 6, 7, 8};
List<Integer> listOfIntegers =
new ArrayList<>(Arrays.asList(intArray));

List<Integer> serialStorage = new ArrayList<>();

System.out.println("Serial stream:");
listOfIntegers
.stream()
// It uses a stateful lambda expression. This should not be done!
.map(e -> {
serialStorage.add(e);
return e;
})
.forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");

serialStorage
.stream()
.forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");

System.out.println("Parallel stream:");
List<Integer> parallelStorage = Collections.synchronizedList(
new ArrayList<>());
listOfIntegers
.parallelStream()
// It uses a stateful lambda expression. This should not be done!
.map(e -> {
parallelStorage.add(e);
return e;
})
.forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");

parallelStorage
.stream()
.forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");
}

The lambda expression e -> { parallelStorage.add(e); return e; } is a stateful lambda expression. Its result can vary every time the code is run. This example prints the following:

Serial stream:
8 7 6 5 4 3 2 1
8 7 6 5 4 3 2 1
Parallel stream:
8 7 6 5 4 3 2 1
1 3 6 2 4 5 8 7

The operation forEachOrdered processes elements in the order specified by the stream, regardless of whether the stream is executed in serial or parallel. However, the point to be kept in mind is when a stream is executed in parallel, the map operation processes elements of the streams specified by the Java runtime and compiler. Consequently, the order in which the lambda expression e -> { parallelStorage.add(e); return e; } adds elements to the List parallelStorage can vary every time the code is run. For deterministic and predictable results, ensure that lambda expression parameters in stream operations are not stateful.

Note : One of the most important feature of streams is that they can be operated only once i.e. reusability of streams is not allowed.

References :

https://docs.oracle.com/javase/tutorial/