Stream Operations

Parallel Streams

For CPU-intensive operations on large datasets, parallel streams automatically split work across multiple cores. Understanding when parallelism helps (and when it hurts) is essential for writing performant stream code.

Creating Parallel Streams

Convert any stream to parallel processing with a single method call.

Create.java 
import java.util.*;
import java.util.stream.*;

public class Create {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> fromParallelStream = nums.parallelStream()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("parallelStream: " + fromParallelStream);

        List<Integer> fromParallel = nums.stream()
                .parallel()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("stream().parallel(): " + fromParallel);

        boolean isParallel = nums.parallelStream().isParallel();
        System.out.println("isParallel: " + isParallel);

    }
}

import java.util.*;
import java.util.stream.*;

public class Create {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> fromParallelStream = nums.parallelStream()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("parallelStream: " + fromParallelStream);

        List<Integer> fromParallel = nums.stream()
                .parallel()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("stream().parallel(): " + fromParallel);

        boolean isParallel = nums.parallelStream().isParallel();
        System.out.println("isParallel: " + isParallel);

    }
}

import java.util.*;
import java.util.stream.*;

public class Create {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> fromParallelStream = nums.parallelStream()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("parallelStream: " + fromParallelStream);

        List<Integer> fromParallel = nums.stream()
                .parallel()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("stream().parallel(): " + fromParallel);

        boolean isParallel = nums.parallelStream().isParallel();
        System.out.println("isParallel: " + isParallel);

    }
}
parallel stream A stream that processes elements concurrently across multiple threads using the fork/join framework.
spliterator The mechanism that divides stream data into chunks for parallel processing; some data structures split better than others.

Performance Considerations

Parallel streams have overhead. They help with large datasets and CPU-bound work, but can hurt performance otherwise.

Performance.java 
import java.util.*;
import java.util.stream.*;

public class Performance {
    private static int work(int n) {
        int result = 0;
        for (int i = 1; i <= 5; i++) {
            result += n * i;
        }
        return result;
    }

    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> sequential = nums.stream()
                .map(Performance::work)
                .collect(Collectors.toList());
        System.out.println("sequential size: " + sequential.size());

        List<Integer> parallel = nums.parallelStream()
                .map(Performance::work)
                .collect(Collectors.toList());
        System.out.println("parallel size: " + parallel.size());

        boolean sameResults = sequential.equals(parallel);
        System.out.println("sameResults: " + sameResults);
    }
}

import java.util.*;
import java.util.stream.*;

public class Performance {
    private static int work(int n) {
        int result = 0;
        for (int i = 1; i <= 5; i++) {
            result += n * i;
        }
        return result;
    }

    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> sequential = nums.stream()
                .map(Performance::work)
                .collect(Collectors.toList());
        System.out.println("sequential size: " + sequential.size());

        List<Integer> parallel = nums.parallelStream()
                .map(Performance::work)
                .collect(Collectors.toList());
        System.out.println("parallel size: " + parallel.size());

        boolean sameResults = sequential.equals(parallel);
        System.out.println("sameResults: " + sameResults);
    }
}

import java.util.*;
import java.util.stream.*;

public class Performance {
    private static int work(int n) {
        int result = 0;
        for (int i = 1; i <= 5; i++) {
            result += n * i;
        }
        return result;
    }

    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> sequential = nums.stream()
                .map(Performance::work)
                .collect(Collectors.toList());
        System.out.println("sequential size: " + sequential.size());

        List<Integer> parallel = nums.parallelStream()
                .map(Performance::work)
                .collect(Collectors.toList());
        System.out.println("parallel size: " + parallel.size());

        boolean sameResults = sequential.equals(parallel);
        System.out.println("sameResults: " + sameResults);
    }
}

Ordering in Parallel Streams

Parallel streams may process elements out of order. Use forEachOrdered when order matters for output.

Order.java 
import java.util.*;
import java.util.stream.*;

public class Order {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> collected = nums.parallelStream()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("collect: " + collected);

        System.out.print("forEachOrdered: ");
        nums.parallelStream().forEachOrdered(n -> System.out.print(n + " "));
        System.out.println();

        List<Integer> result = nums.parallelStream()
                .map(n -> n * 2)
                .collect(Collectors.toList());
        System.out.println("collect: " + result);

    }
}

import java.util.*;
import java.util.stream.*;

public class Order {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> collected = nums.parallelStream()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("collect: " + collected);

        System.out.print("forEachOrdered: ");
        nums.parallelStream().forEachOrdered(n -> System.out.print(n + " "));
        System.out.println();

        List<Integer> result = nums.parallelStream()
                .map(n -> n * 2)
                .collect(Collectors.toList());
        System.out.println("collect: " + result);

    }
}

import java.util.*;
import java.util.stream.*;

public class Order {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> collected = nums.parallelStream()
                .map(n -> n)
                .collect(Collectors.toList());
        System.out.println("collect: " + collected);

        System.out.print("forEachOrdered: ");
        nums.parallelStream().forEachOrdered(n -> System.out.print(n + " "));
        System.out.println();

        List<Integer> result = nums.parallelStream()
                .map(n -> n * 2)
                .collect(Collectors.toList());
        System.out.println("collect: " + result);

    }
}
encounter order The defined sequence of elements in a stream; parallel operations may process out of order but can preserve order in results.

Avoiding Shared Mutable State

Shared mutable state in parallel streams causes race conditions. Keep operations stateless.

SharedState.java 
import java.util.*;
import java.util.stream.*;

public class SharedState {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> sideEffectResult = Collections.synchronizedList(new ArrayList<>());
        nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .forEach(sideEffectResult::add);
        Collections.sort(sideEffectResult);
        System.out.println("sideEffectResult size: " + sideEffectResult.size());

        List<Integer> correctResult = nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .collect(Collectors.toList());
        System.out.println("correctResult size: " + correctResult.size());

    }
}

import java.util.*;
import java.util.stream.*;

public class SharedState {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> sideEffectResult = Collections.synchronizedList(new ArrayList<>());
        nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .forEach(sideEffectResult::add);
        Collections.sort(sideEffectResult);
        System.out.println("sideEffectResult size: " + sideEffectResult.size());

        List<Integer> correctResult = nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .collect(Collectors.toList());
        System.out.println("correctResult size: " + correctResult.size());

    }
}

import java.util.*;
import java.util.stream.*;

public class SharedState {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        List<Integer> sideEffectResult = Collections.synchronizedList(new ArrayList<>());
        nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .forEach(sideEffectResult::add);
        Collections.sort(sideEffectResult);
        System.out.println("sideEffectResult size: " + sideEffectResult.size());

        List<Integer> correctResult = nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .collect(Collectors.toList());
        System.out.println("correctResult size: " + correctResult.size());

    }
}
thread safety Code that behaves correctly when accessed by multiple threads simultaneously; parallel streams require stateless operations.

Combiners for Parallel Reduce

When reducing in parallel, provide a combiner to merge partial results from different threads.

Combiners.java 
import java.util.*;
import java.util.stream.*;

public class Combiners {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        int sum = nums.parallelStream()
                .reduce(0, Integer::sum, Integer::sum);
        System.out.println("sum: " + sum);

        List<Integer> evens = nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .collect(
                        ArrayList::new,
                        ArrayList::add,
                        ArrayList::addAll // combiner
                );
        System.out.println("evens count: " + evens.size());

        Map<Integer, List<Integer>> byMod = nums.parallelStream()
                .collect(Collectors.groupingBy(n -> n % 5, TreeMap::new, Collectors.toList()));
        System.out.println("byMod: " + byMod.keySet());

    }
}

import java.util.*;
import java.util.stream.*;

public class Combiners {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        int sum = nums.parallelStream()
                .reduce(0, Integer::sum, Integer::sum);
        System.out.println("sum: " + sum);

        List<Integer> evens = nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .collect(
                        ArrayList::new,
                        ArrayList::add,
                        ArrayList::addAll // combiner
                );
        System.out.println("evens count: " + evens.size());

        Map<Integer, List<Integer>> byMod = nums.parallelStream()
                .collect(Collectors.groupingBy(n -> n % 5, TreeMap::new, Collectors.toList()));
        System.out.println("byMod: " + byMod.keySet());

    }
}

import java.util.*;
import java.util.stream.*;

public class Combiners {
    public static void main(String[] args) {
        int rangeEnd = ;
        List<Integer> nums = IntStream.range(1, rangeEnd).boxed().toList();

        int sum = nums.parallelStream()
                .reduce(0, Integer::sum, Integer::sum);
        System.out.println("sum: " + sum);

        List<Integer> evens = nums.parallelStream()
                .filter(n -> n % 2 == 0)
                .collect(
                        ArrayList::new,
                        ArrayList::add,
                        ArrayList::addAll // combiner
                );
        System.out.println("evens count: " + evens.size());

        Map<Integer, List<Integer>> byMod = nums.parallelStream()
                .collect(Collectors.groupingBy(n -> n % 5, TreeMap::new, Collectors.toList()));
        System.out.println("byMod: " + byMod.keySet());

    }
}

Exercise: Practical.java

Compare sequential vs parallel performance for computing statistics on a large dataset