Java Module 11 – Streams and stream gatherers

Java parallel streams: when they help

Measure Java parallel streams on JDK 25: the common ForkJoinPool, splitting and ordering costs, cases where parallel is slower, and data races to avoid.

Advanced
24 minutes
Examples run with Java 25.0.4 and Python 3.14.8
By MySmartCoPilot

Free: the whole lesson. With a pass: the model answers to its 2 interview questions.

What you will learn

Run pipelines in parallel and measure the speed-up honestly
Explain the common ForkJoinPool, splitting and ordering costs
Avoid shared mutable state in parallel pipelines

Before you start

Java primitive streams and summary statistics

What you’ll learn · Before you start

What you will learn

Run pipelines in parallel and measure the speed-up honestly
Explain the common ForkJoinPool, splitting and ordering costs
Avoid shared mutable state in parallel pipelines

Before you start

Java primitive streams and summary statistics

On this page

A parallel stream splits its source into parts, processes the parts at the same time on several threads, and joins the partial results into one answer. You ask for one with collection.parallelStream() or stream.parallel(), and the threads come from the common ForkJoinPool, shared by the whole JVM, plus the thread that started the stream. It helps only when there is enough computing to share: many elements or expensive work per element, a source that splits into equal parts, such as an array, an ArrayList or IntStream.range, results that are cheap to join, no waiting on input or output, and no shared mutable state. Otherwise it is slower or wrong. This lesson measures each case on an 18-core machine: work that ran several times faster, inputs where parallel lost or gained nothing, a shared list that dropped most of its elements, and blocking calls that held up an unrelated task.

What parallel() does

What a parallel stream does with its source

Text description of the diagram

The diagram is one column of five steps, read from top to bottom.

The source: 1,000,000 elements in an ArrayList, an array or an IntStream.range.
Split: Spliterator.trySplit() halves the source, again and again, into small parts. An arrow labelled "parts of about equal size" leads to the next step.
Work on the parts at the same time: the calling thread and the workers of the common ForkJoinPool, 17 of them on the machine of the recorded runs, each process parts. An arrow labelled "one partial result per part" leads to the next step.
Join: the partial sums, lists or maps are combined, in encounter order where it matters.
An oval: one result, the same as the sequential one.

The program below sums a function of ten million numbers in parallel and records the name of every thread that did some of the work, then checks the answer against the sequential one:

Threads.java Java · Threads.java

import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.LongStream;

class Threads {
    /** "ForkJoinPool.commonPool-worker-3" becomes "a common-pool worker"; "main" stays "main". */
    static String kind(String threadName) {
        return threadName.startsWith("ForkJoinPool.commonPool-worker-") ? "common-pool workers" : threadName;
    }

    static void main() {
        IO.println("processors: " + Runtime.getRuntime().availableProcessors()
                + ", common pool parallelism: " + ForkJoinPool.getCommonPoolParallelism());

        Set<String> names = ConcurrentHashMap.newKeySet();
        long parallelSum = LongStream.rangeClosed(1, 10_000_000).parallel()
                .peek(x -> names.add(Thread.currentThread().getName()))   // peek only to see the threads
                .map(x -> x * x % 1_000_000_007)
                .sum();
        long sequentialSum = LongStream.rangeClosed(1, 10_000_000).map(x -> x * x % 1_000_000_007).sum();
        IO.println("same sum: " + (parallelSum == sequentialSum) + " (" + "%,d".formatted(parallelSum) + ")");
        Set<String> kinds = new TreeSet<>();
        names.forEach(n -> kinds.add(kind(n)));
        IO.println("threads that did the work: " + kinds + ", " + (names.size() - 1) + " workers besides main");

        // The last parallel() or sequential() call decides for the whole pipeline.
        Set<String> last = ConcurrentHashMap.newKeySet();
        LongStream.rangeClosed(1, 1_000_000).parallel()
                .map(x -> x * x % 1_000_000_007)
                .sequential()
                .forEach(x -> last.add(Thread.currentThread().getName()));
        IO.println("parallel() then sequential(): " + last);
    }
}

Which threads run a parallel stream, recorded Python · threads_session.py

from session import machine, run, version

version()
machine()
run("""
java Threads.java
""")

Output

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java Threads.java
processors: 18, common pool parallelism: 17
same sum: true (4,993,511,036,287,577)
threads that did the work: [common-pool workers, main], 17 workers besides main
parallel() then sequential(): [main]

This output changes from run to run: How many workers pick up a part depends on the machine and on timing; the sum and the last line are the same on every run.

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 threads_session.py

On the recorded machine the common pool had 17 workers for 18 processors, one fewer, because the thread that calls the terminal operation works on parts too, so all 18 processors were busy. The sum was the same as the sequential one, as the java.util.stream documentation requires: apart from operations that are explicitly nondeterministic, such as findAny, running in parallel must not change the result. The last line shows a rule that surprises people: parallel() and sequential() are not stage-by-stage switches. The last call decides for the whole pipeline, so .parallel().map(...).sequential().forEach(...) ran everything on main.

Collection.parallelStream() returns a stream that is only “possibly parallel”, in its Javadoc’s words: a collection may return a sequential one. isParallel() tells you which you got.

Measuring the speed-up honestly

A fair comparison runs both versions on the same input, throws the first runs away while the JIT compiler is still compiling, keeps the best of several timed runs, and checks that both give the same answer. The program below does that for six kinds of work and prints the hardware with the times:

Speedup.java Java · Speedup.java

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.function.LongSupplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;
import java.util.stream.Stream;

class Speedup {
    static final long MOD = 1_000_000_007;

    static boolean isPrime(int n) {
        if (n < 2) return false;
        for (int d = 2; (long) d * d <= n; d++) if (n % d == 0) return false;
        return true;
    }

    /** Best time of `rounds` runs after `warmUp` untimed ones; every run must give the expected answer. */
    static double bestMillis(LongSupplier work, long expected, int warmUp, int rounds) {
        long best = Long.MAX_VALUE;
        for (int i = 0; i < warmUp + rounds; i++) {
            long start = System.nanoTime();
            long answer = work.getAsLong();
            long nanos = System.nanoTime() - start;
            if (answer != expected) throw new AssertionError("wrong answer " + answer);
            if (i >= warmUp) best = Math.min(best, nanos);
        }
        return best / 1e6;
    }

    static void compare(String what, LongSupplier sequential, LongSupplier parallel, int warmUp, int rounds) {
        long expected = sequential.getAsLong();
        double s = bestMillis(sequential, expected, warmUp, rounds);
        double p = bestMillis(parallel, expected, warmUp, rounds);
        double ratio = s / p;
        String verdict = ratio >= 2 ? "faster" : ratio <= 0.9 ? "slower" : "about the same";
        IO.println("%-36s %9.3f ms %9.3f ms  %5.1f×  parallel %s".formatted(what, s, p, ratio, verdict));
    }

    static void main() {
        IO.println("%-36s %12s %12s  %6s".formatted("work", "sequential", "parallel", "speed-up"));
        compare("x² mod 1,000,000,007, 10,000,000 longs",
                () -> LongStream.rangeClosed(1, 10_000_000).map(x -> x * x % MOD).sum(),
                () -> LongStream.rangeClosed(1, 10_000_000).parallel().map(x -> x * x % MOD).sum(), 5, 10);
        compare("primes below 5,000,000",
                () -> IntStream.range(0, 5_000_000).filter(Speedup::isPrime).count(),
                () -> IntStream.range(0, 5_000_000).parallel().filter(Speedup::isPrime).count(), 2, 4);
        compare("x² mod 1,000,000,007, 1,000 longs",
                () -> LongStream.rangeClosed(1, 1_000).map(x -> x * x % MOD).sum(),
                () -> LongStream.rangeClosed(1, 1_000).parallel().map(x -> x * x % MOD).sum(), 50, 50);
        compare("Stream.iterate + limit, 1,000,000",
                () -> Stream.iterate(1L, x -> x + 1).limit(1_000_000).mapToLong(x -> x * x % MOD).sum(),
                () -> Stream.iterate(1L, x -> x + 1).parallel().limit(1_000_000).mapToLong(x -> x * x % MOD).sum(), 5, 10);
        List<Long> array = LongStream.rangeClosed(1, 1_000_000).boxed().collect(Collectors.toCollection(ArrayList::new));
        List<Long> linked = new LinkedList<>(array);
        compare("ArrayList<Long>, 1,000,000",
                () -> array.stream().mapToLong(x -> x * x % MOD).sum(),
                () -> array.parallelStream().mapToLong(x -> x * x % MOD).sum(), 5, 10);
        compare("LinkedList<Long>, 1,000,000",
                () -> linked.stream().mapToLong(x -> x * x % MOD).sum(),
                () -> linked.parallelStream().mapToLong(x -> x * x % MOD).sum(), 5, 10);
    }
}

Sequential and parallel times for six kinds of work, measured Python · speedup_session.py

from session import machine, run, version

version()
machine()
run("""
java Speedup.java
""")

Output

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java Speedup.java
work                                   sequential     parallel  speed-up
x² mod 1,000,000,007, 10,000,000 longs     9.604 ms     1.318 ms    7.3×  parallel faster
primes below 5,000,000                 654.589 ms    77.078 ms    8.5×  parallel faster
x² mod 1,000,000,007, 1,000 longs        0.010 ms     0.021 ms    0.5×  parallel slower
Stream.iterate + limit, 1,000,000        8.562 ms     6.399 ms    1.3×  parallel about the same
ArrayList<Long>, 1,000,000               3.195 ms     0.911 ms    3.5×  parallel faster
LinkedList<Long>, 1,000,000              3.226 ms     2.633 ms    1.2×  parallel about the same

This output changes from run to run: The times change from run to run with the load on the computer; which version is faster, slower or about the same holds on every run.

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 speedup_session.py

The program calls parallel “faster” only from twice the speed, “slower” below 0.9 of it, and “about the same” in between. Read the rows in pairs:

Large input, cheap work, a source that splits well. Ten million numbers from LongStream.rangeClosed, each squared and reduced modulo 1,000,000,007, ran several times faster in parallel. Each element costs only a few nanoseconds, but there are enough of them to keep every core busy.
Expensive work per element. Testing every number below five million for primality by trial division took hundreds of milliseconds sequentially and a fraction of that in parallel, because the cost of splitting and joining is tiny next to the work.
Small input. A thousand elements took microseconds sequentially, and the parallel version was slower: handing parts to other threads and waiting for them costs more than the whole job.
A source that cannot split. Stream.iterate produces each element from the one before, so it cannot be cut into halves without walking it; the stream copies batches of it into arrays first, and limit on an ordered parallel stream adds work of its own. Parallel gained little or nothing.
The same million numbers, two lists. An ArrayList splits by index into exact halves and ran several times faster in parallel. A LinkedList has to be walked from its head to find a middle, and parallel gained little or nothing. The Spliterator Javadoc says it directly: a source that splits in “a highly imbalanced or inefficient manner” is unlikely to benefit from parallelism.

The table measures one machine, busy with other work at the time. A smaller laptop, a container limited to two processors or a loaded server can turn “faster” into “about the same”, so measure on the hardware that will run the code. For results you want to publish, use a benchmark harness such as JMH, which controls warm-up, dead-code elimination and run-to-run noise far better than a loop with System.nanoTime().

Order has a price

Most streams are ordered: a List, an array and a range all have an encounter order, and operations such as toList(), findFirst() and limit(n) must respect it, even in parallel. Respecting it means threads sometimes wait for each other or keep results they may throw away:

Ordering.java Java · Ordering.java

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.function.LongSupplier;
import java.util.stream.IntStream;

class Ordering {
    static boolean isPrime(int n) {
        if (n < 2) return false;
        for (int d = 2; (long) d * d <= n; d++) if (n % d == 0) return false;
        return true;
    }

    /** Best time of `rounds` runs, after 3 untimed runs that let the JIT compiler finish. */
    static double bestMillis(LongSupplier work, int rounds) {
        long best = Long.MAX_VALUE;
        for (int i = 0; i < 3 + rounds; i++) {
            long start = System.nanoTime();
            work.getAsLong();
            if (i >= 3) best = Math.min(best, System.nanoTime() - start);
        }
        return best / 1e6;
    }

    static void main() {
        // forEach on a parallel stream runs the action in whatever order the threads reach the elements.
        List<Integer> seen = Collections.synchronizedList(new ArrayList<>());
        IntStream.rangeClosed(1, 12).parallel().forEach(seen::add);
        IO.println("forEach:        " + seen);
        // forEachOrdered waits, so that the action sees the elements in their encounter order.
        List<Integer> ordered = new ArrayList<>();
        IntStream.rangeClosed(1, 12).parallel().forEachOrdered(ordered::add);
        IO.println("forEachOrdered: " + ordered);
        // A collected result keeps the encounter order on any stream.
        IO.println("toList():       " + IntStream.rangeClosed(1, 12).parallel().boxed().toList());

        // findFirst must return the first match in order; findAny may return any match.
        IO.println("findFirst prime above 4,900,000: " + IntStream.range(0, 5_000_000).parallel()
                .filter(n -> n > 4_900_000 && isPrime(n)).findFirst().getAsInt());
        double first = bestMillis(() -> IntStream.range(0, 5_000_000).parallel()
                .filter(n -> n > 4_900_000 && isPrime(n)).findFirst().getAsInt(), 12);
        double any = bestMillis(() -> IntStream.range(0, 5_000_000).parallel()
                .filter(n -> n > 4_900_000 && isPrime(n)).findAny().getAsInt(), 12);
        IO.println("findFirst %.2f ms, findAny %.2f ms; findAny took less time: %s".formatted(first, any, any < first));

        // limit(n) on an ordered parallel stream must keep the first n elements; unordered() lets it keep any n.
        double limitOrdered = bestMillis(() -> IntStream.range(0, 5_000_000).parallel()
                .filter(Ordering::isPrime).limit(100_000).count(), 6);
        double limitUnordered = bestMillis(() -> IntStream.range(0, 5_000_000).parallel().unordered()
                .filter(Ordering::isPrime).limit(100_000).count(), 6);
        IO.println("limit(100,000) of the primes: ordered %.1f ms, unordered() %.1f ms; unordered() took less time: %s"
                .formatted(limitOrdered, limitUnordered, limitUnordered < limitOrdered));
    }
}

forEach against forEachOrdered, findFirst against findAny, and limit with and without order Python · ordering_session.py

from session import machine, run, version

version()
machine()
run("""
java Ordering.java
""")

Output

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java Ordering.java
forEach:        [8, 9, 7, 11, 12, 10, 1, 4, 5, 2, 3, 6]
forEachOrdered: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
toList():       [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
findFirst prime above 4,900,000: 4900001
findFirst 0.78 ms, findAny 0.29 ms; findAny took less time: true
limit(100,000) of the primes: ordered 30.3 ms, unordered() 15.1 ms; unordered() took less time: true

This output changes from run to run: The order in which forEach meets the elements and all the times change from run to run; the ordered results, and which version took less time, are the same on every run.

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 ordering_session.py

forEach ignores order on a parallel stream; the twelve numbers arrived scrambled. forEachOrdered delivered them in order, but only by holding back results that were ready early. toList() kept the order without that cost to your code, because each part fills its own piece of the list and the pieces are joined in order.
findFirst must return the first match in encounter order, so threads that find a later match wait for the earlier parts. findAny may return any match and stops as soon as one thread has one, and it took less time. Use it when any match will do, such as “is there an overdue invoice?”, and remember that its answer may change from run to run.
limit(n) on an ordered parallel stream must keep exactly the first n elements, which the limit Javadoc calls potentially “quite expensive”. After unordered() it may keep any n, and it took less time. unordered() only helps where any n elements are an acceptable answer.

Shared mutable state: the race you cannot see in tests

Lambdas in a parallel pipeline run on many threads at once. A lambda that changes something outside itself, such as an ArrayList or a counter, is changed by several threads without coordination:

Race.java Java · Race.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;

class Race {
    static final int N = 1_000_000;

    static void main() {
        // Five tries at filling one shared ArrayList from a parallel forEach.
        int correct = 0;
        for (int round = 1; round <= 5; round++) {
            List<Integer> shared = new ArrayList<>();
            try {
                IntStream.range(0, N).parallel().forEach(shared::add);
                if (shared.size() == N) correct++;
                IO.println("round " + round + ": the list holds %,d of %,d numbers".formatted(shared.size(), N));
            } catch (RuntimeException e) {
                IO.println("round " + round + ": " + e.getClass().getName());
            }
        }
        IO.println("rounds with all %,d numbers: %d of 5".formatted(N, correct));

        // A shared counter loses updates the same way: counter[0]++ is a read, an add and a write.
        long[] counter = {0};
        IntStream.range(0, N).parallel().forEach(i -> counter[0]++);
        IO.println("long[] counter: %,d (lost %,d)".formatted(counter[0], N - counter[0]));

        // Correct: a thread-safe counter, or better, no shared state at all.
        AtomicLong atomic = new AtomicLong();
        IntStream.range(0, N).parallel().forEach(i -> atomic.incrementAndGet());
        IO.println("AtomicLong: %,d".formatted(atomic.get()));
        List<Integer> collected = IntStream.range(0, N).parallel().boxed().toList();
        IO.println("toList(): %,d numbers, in order: %s".formatted(collected.size(),
                IntStream.range(0, N).allMatch(i -> collected.get(i) == i)));
    }
}

Adding to a shared ArrayList and a shared counter from a parallel forEach Python · race_session.py

from session import machine, run, version

version()
machine()
run("""
java Race.java
""")

Output

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java Race.java
round 1: the list holds 271,985 of 1,000,000 numbers
round 2: the list holds 227,303 of 1,000,000 numbers
round 3: java.lang.ArrayIndexOutOfBoundsException
round 4: java.lang.ArrayIndexOutOfBoundsException
round 5: java.lang.ArrayIndexOutOfBoundsException
rounds with all 1,000,000 numbers: 0 of 5
long[] counter: 437,430 (lost 562,570)
AtomicLong: 1,000,000
toList(): 1,000,000 numbers, in order: true

This output changes from run to run: A data race gives a different wrong answer on every run; the correct versions print the same on every run.

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 race_session.py

Not one of the five rounds kept all million numbers. ArrayList.add reads the size, writes the element and stores the new size; two threads that read the same size write into the same slot and one element is lost, and when one thread grows the internal array while another writes, the write lands outside it and throws ArrayIndexOutOfBoundsException. The counter lost most of its updates for the same reason: counter[0]++ is a read, an addition and a write. A sequential test never shows either problem, and a small parallel test may pass by luck.

AtomicLong made the count correct, but every thread now competes for one memory location, which undoes much of the gain. The real fix is to have no shared state: let the stream build the result. toList(), collect, sum() and count() give each part its own container and join them, so they are correct and fast, and toList() even kept the order. The java.util.stream documentation uses almost this example to explain why forEach that fills a collection is an unnecessary side effect.

Blocking work starves the common pool

The common pool is small on purpose: about one thread per processor, sized for computing. A parallel stream whose lambdas wait, for a remote call, a database query or a file, keeps those threads asleep, and everything else that uses the pool waits too. That includes CompletableFuture.supplyAsync without an executor, which its Javadoc says runs on ForkJoinPool.commonPool():

Blocking.java Java · Blocking.java

import java.util.concurrent.CompletableFuture;
import java.util.stream.Gatherers;
import java.util.stream.IntStream;

class Blocking {
    /** Stands in for a call that waits on the network for 400 ms. */
    static int call(int i) {
        try {
            Thread.sleep(400);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
        return i;
    }

    static long millisSince(long start) {
        return (System.nanoTime() - start) / 1_000_000;
    }

    static void main() throws Exception {
        // 36 blocking calls in a parallel stream, on its own thread, so that main can try the pool meanwhile.
        long start = System.nanoTime();
        Thread batch = Thread.ofPlatform().start(() -> IntStream.range(0, 36).parallel().map(Blocking::call).sum());
        Thread.sleep(100);

        // supplyAsync without an executor also runs on the common pool: it has to wait for a free worker.
        long asked = System.nanoTime();
        int answer = CompletableFuture.supplyAsync(() -> 6 * 7).get();
        long waited = millisSince(asked);
        batch.join();
        long parallelMillis = millisSince(start);

        // The same 36 calls with mapConcurrent(36): virtual threads, and the common pool stays free.
        start = System.nanoTime();
        long count = IntStream.range(0, 36).boxed().gather(Gatherers.mapConcurrent(36, Blocking::call)).count();
        long concurrentMillis = millisSince(start);

        IO.println("a tiny supplyAsync task waited at least 100 ms: " + (waited >= 100)
                + " (about %,d ms, answer %d)".formatted(Math.round(waited / 100.0) * 100, answer));
        IO.println("36 calls of 400 ms, parallel stream: about %,d ms".formatted(Math.round(parallelMillis / 100.0) * 100));
        IO.println("36 calls of 400 ms, mapConcurrent(36): about %,d ms (%d results)".formatted(Math.round(concurrentMillis / 100.0) * 100, count));
    }
}

36 blocking calls in a parallel stream, and a tiny task that needed the same pool Python · blocking_session.py

from session import machine, run, version

version()
machine()
run("""
java Blocking.java
""")

Output

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java Blocking.java
a tiny supplyAsync task waited at least 100 ms: true (about 300 ms, answer 42)
36 calls of 400 ms, parallel stream: about 800 ms
36 calls of 400 ms, mapConcurrent(36): about 400 ms (36 results)

This output changes from run to run: The waits depend on when the pool's threads become free; that the small task had to wait, and the mapConcurrent time, hold on every run.

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 blocking_session.py

A task that multiplies 6 by 7 waited hundreds of milliseconds, because every worker of the pool was asleep inside a parallel stream. The 36 calls themselves ran 18 at a time, one per thread, so they took two rounds of 400 ms; on a machine with 4 processors they would take nine. The ForkJoinPool Javadoc warns about this: the pool tries to keep enough threads active, but it guarantees no such adjustment “in the face of blocked I/O”. For work that waits, use virtual threads: Gatherers.mapConcurrent(36, ...) finished in one round of 400 ms and left the common pool free, as the stream gatherers lesson shows. Running a parallel stream inside a ForkJoinPool of your own is sometimes suggested as a workaround, but the documentation of streams does not promise which pool a stream uses, so do not build on it.

Before you write parallel()

The measurements above come down to six points to check before adding parallel() to a pipeline:

What helps a parallel stream, and what hurts it
Criterion	Helps parallel	Hurts parallel
How much work	Millions of cheap elements, or thousands of expensive ones	Hundreds of cheap elements
The source	Arrays, ArrayList, IntStream.range	LinkedList, Stream.iterate, an Iterator turned into a stream
Each element	Computes, with no waiting	Waits on the network, a database or a disk
Joining results	sum, count, toList, toMap with a merge function	Building one big object step by step, forEachOrdered
Order	Does not matter, or only in the collected result	findFirst, limit and skip on ordered data
Shared state	None: every lambda only reads its input	Lambdas that add to a list, a map or a counter

If a pipeline has something from both columns, measure before you decide, and keep the sequential version when the gain is small: a parallel stream also takes cores from every other request that the same JVM is serving.

Your turn: a parallel word count

The method below counts words in parallel. It has three bugs, and only one of them is about threads. Find them, then write a version that gives the same counts as a sequential one on every run and is faster than it:

WordCountBuggy.java: the version to fix Java · WordCountBuggy.java

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

class WordCountBuggy {
    /** The version to fix: it is neither correct nor safe, on a parallel stream or off it. */
    static Map<String, Integer> count(List<String> lines) {
        Map<String, Integer> counts = new HashMap<>();
        lines.parallelStream()
                .flatMap(line -> Arrays.stream(line.toLowerCase().split("\\W+")))
                .forEach(word -> counts.merge(word, 1, Integer::sum));
        return counts;
    }
}

WordCount.java: one solution, with its checks Java · WordCount.java

import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.SplittableRandom;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;

class WordCount {
    /** Anything that is not a letter, a combining mark (such as the vowel signs of Hindi), a digit or an apostrophe. */
    static final Pattern NOT_WORD = Pattern.compile("[^\\p{L}\\p{M}\\p{N}']+");

    static Stream<String> words(String line) {
        return NOT_WORD.splitAsStream(line.toLowerCase(Locale.ROOT)).filter(w -> !w.isEmpty());
    }

    static Map<String, Long> sequential(List<String> lines) {
        return lines.stream().flatMap(WordCount::words)
                .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
    }

    /** The fix: no shared map. Each part of the stream counts into its own map, and the maps are merged. */
    static Map<String, Long> parallel(List<String> lines) {
        return lines.parallelStream().flatMap(WordCount::words)
                .collect(Collectors.toMap(Function.identity(), w -> 1L, Long::sum));
    }

    /** Also correct: one ConcurrentHashMap that every thread updates. */
    static Map<String, Long> parallelConcurrent(List<String> lines) {
        return lines.parallelStream().flatMap(WordCount::words)
                .collect(Collectors.groupingByConcurrent(Function.identity(), Collectors.counting()));
    }

    /** 100,000 made-up lines of 12 words each from a 3,000-word vocabulary; a few words are far more common. */
    static List<String> text() {
        SplittableRandom random = new SplittableRandom(2026);
        String[] vocabulary = new String[3_000];
        for (int i = 0; i < vocabulary.length; i++) {
            StringBuilder w = new StringBuilder();
            for (int j = 2 + random.nextInt(8); j > 0; j--) w.append((char) ('a' + random.nextInt(26)));
            vocabulary[i] = w.toString();
        }
        List<String> lines = new ArrayList<>(100_000);
        for (int l = 0; l < 100_000; l++) {
            StringBuilder line = new StringBuilder(l % 50 == 0 ? "\"" : "");   // some lines start with a quote mark
            for (int w = 0; w < 12; w++) {
                String word = vocabulary[(int) (Math.pow(random.nextDouble(), 3) * vocabulary.length)];
                line.append(random.nextInt(10) == 0 ? word.toUpperCase(Locale.ROOT) : word)
                        .append(random.nextInt(8) == 0 ? ", " : " ");
            }
            lines.add(line.toString());
        }
        return lines;
    }

    static double bestMillis(Supplier<Map<String, Long>> count, Map<String, Long> expected) {
        long best = Long.MAX_VALUE;
        for (int i = 0; i < 6; i++) {
            long start = System.nanoTime();
            Map<String, Long> result = count.get();
            long nanos = System.nanoTime() - start;
            if (!result.equals(expected)) throw new AssertionError("a run gave different counts");
            if (i >= 2) best = Math.min(best, nanos);
        }
        return best / 1e6;
    }

    static void check(String what, boolean ok) {
        if (!ok) throw new AssertionError("FAILED: " + what);
        IO.println("ok   " + what);
    }

    static void main() {
        List<String> small = List.of("\"The cat and the hat.\"", "THE END", "café नमस्ते café, it's");
        IO.println("small text: " + new java.util.TreeMap<>(parallel(small)));
        check("upper and lower case count as one word", parallel(small).get("the") == 3);
        check("no empty word from a leading quote mark", !parallel(small).containsKey(""));
        check("accents, Hindi vowel signs and apostrophes stay inside words",
                parallel(small).get("café") == 2 && parallel(small).get("नमस्ते") == 1 && parallel(small).get("it's") == 1);

        List<String> lines = text();
        Map<String, Long> expected = sequential(lines);
        check("1,200,000 words counted", expected.values().stream().mapToLong(Long::longValue).sum() == 1_200_000);
        boolean same = true;
        for (int run = 0; run < 50; run++) same &= parallel(lines).equals(expected);
        check("50 parallel runs give the sequential counts", same);

        double s = bestMillis(() -> sequential(lines), expected);
        double p = bestMillis(() -> parallel(lines), expected);
        double c = bestMillis(() -> parallelConcurrent(lines), expected);
        IO.println("sequential %.0f ms; parallel, toMap with merge %.0f ms (%.1f× faster); parallel, groupingByConcurrent %.0f ms"
                .formatted(s, p, s / p, c));
        check("the parallel version is at least twice as fast as the sequential one", s / p >= 2);
        IO.println("all checks passed");
    }
}

The checks, and the times of the solution Python · exercise_session.py

from session import machine, run, version

version()
machine()
run("""
java WordCount.java
""")

Output

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java WordCount.java
small text: {and=1, café=2, cat=1, end=1, hat=1, it's=1, the=3, नमस्ते=1}
ok   upper and lower case count as one word
ok   no empty word from a leading quote mark
ok   accents, Hindi vowel signs and apostrophes stay inside words
ok   1,200,000 words counted
ok   50 parallel runs give the sequential counts
sequential 204 ms; parallel, toMap with merge 21 ms (9.5× faster); parallel, groupingByConcurrent 57 ms
ok   the parallel version is at least twice as fast as the sequential one
all checks passed

This output changes from run to run: The times change from run to run with the load on the computer; the counts and the checks are the same on every run.

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 exercise_session.py

The three bugs, in order of how much damage they do:

A shared HashMap written from many threads. merge on a HashMap is not thread-safe, so counts are lost and the map’s internal structure can break. The solution lets the stream build the map: toMap(word -> word, word -> 1L, Long::sum) gives each part its own map and merges the maps with Long::sum.
split("\\W+") and an empty first word. A line that starts with a quote mark splits into "" first, so the buggy version counts an empty word. By default \W also treats every character except ASCII letters, digits and _ as a separator, é and every Hindi letter included. The solution splits on anything that is not a letter, a combining mark, a digit or an apostrophe, and drops empty strings. The combining marks matter: नमस्ते is written with a virama and a vowel sign, which are marks, not letters, and without \p{M} the word would fall apart.
toLowerCase() without a locale. It uses the computer’s default locale; in Turkish, "TITLE".toLowerCase() is “tıtle” with a dotless ı. Counting words needs the same answer everywhere, so the solution uses Locale.ROOT.

On the recorded run the solution was several times faster than the sequential count, and also faster than groupingByConcurrent, which puts every thread on one ConcurrentHashMap: with a few very common words, threads keep updating the same entries.

Interview questions

Fresher · Which thread pool runs a parallel stream, and how many threads does it use by default?

A parallel stream runs on the common ForkJoinPool, one pool shared by the whole JVM, and the thread that calls the terminal operation works on parts too. ForkJoinPool.getCommonPoolParallelism() reports its target size; on an 18-processor machine it printed 17, so 18 threads did the work. The same pool runs CompletableFuture.supplyAsync and other async methods that are given no executor, so a parallel stream that blocks on input or output delays all of them. The size can be set with the system property java.util.concurrent.ForkJoinPool.common.parallelism, though the JDK 25 Javadoc discourages it. For work that waits rather than computes, use virtual threads, for example Gatherers.mapConcurrent, instead of a parallel stream.

Model answers open with any pass: ₹79 for 1 week ($2.99), one payment that never renews.

Choose a pass

Fresher to mid

Code review: results = new ArrayList<>(); orders.parallelStream().filter(Order::isLate).forEach(results::add). In production the list is sometimes short, and once the code threw ArrayIndexOutOfBoundsException. Explain both symptoms, give two correct fixes, and say which one you would merge.

Model answer, Pro. Model answers open with any pass; prices are in the notice on this page. First step and follow-ups
Mid to senior

After a release, a report endpoint runs customers.parallelStream().map(this::fetchCreditScore), and under load unrelated endpoints that use CompletableFuture.supplyAsync become slow. Explain the link and propose a fix, with numbers for 2,000 customers, a 150 ms remote call and a 16-core server.

Model answer, Pro. First step and follow-ups

Key takeaways

parallel() splits the source, works on the parts in the common ForkJoinPool plus the calling thread, and joins the results; the last parallel() or sequential() call decides for the whole pipeline.
Parallel pays off for many elements or expensive work on a source that splits evenly; on the recorded machine it was several times faster for ten million numbers and a prime search, and slower for a thousand numbers.
Stream.iterate, LinkedList and other sources that split badly gain little or nothing.
forEachOrdered, findFirst and limit keep encounter order at a cost; findAny and unordered() can be faster when order does not matter.
Never change shared state from a parallel pipeline: a shared ArrayList lost elements or threw, and a shared counter lost updates. Let toList(), collect, sum or count build the result.
Blocking calls in a parallel stream starve every other user of the common pool; use virtual threads for them.
Measure with warm-up, several runs, a check of the answer and the hardware named; use JMH for figures you publish.

References

Package java.util.stream (Java SE 25): Parallelism, Side-effects and Ordering (Oracle)
ForkJoinPool (Java SE 25) (Oracle)
Spliterator (Java SE 25) (Oracle)
Stream (Java SE 25): limit, findFirst and findAny (Oracle)
Collection (Java SE 25): parallelStream (Oracle)
CompletableFuture (Java SE 25) (Oracle)
String (Java SE 25): toLowerCase and locales (Oracle)
Pattern (Java SE 25): predefined character classes (Oracle)
Collectors (Java SE 25): toMap and groupingByConcurrent (Oracle)
Parallelizing Streams (Oracle)

Report a problem with this lesson

In this module: Streams and stream gatherers

Java primitive streams and summary statistics Free lesson, Pro answers: 18 minutes
Stream gatherers: custom intermediate operations Free lesson, Pro answers: 24 minutes
Java parallel streams: when they help Free lesson, Pro answers: 24 minutes
Java stream pitfalls, exceptions and debugging Free lesson, Pro answers: 22 minutes

Your country

Java parallel streams: when they help

What you will learn

Before you start

What you will learn

Before you start

What parallel() does

Measuring the speed-up honestly

Order has a price

Shared mutable state: the race you cannot see in tests

Blocking work starves the common pool

Before you write parallel()

Your turn: a parallel word count

Interview questions

Code review: results = new ArrayList<>(); orders.parallelStream().filter(Order::isLate).forEach(results::add). In production the list is sometimes short, and once the code threw ArrayIndexOutOfBoundsException. Explain both symptoms, give two correct fixes, and say which one you would merge.

After a release, a report endpoint runs customers.parallelStream().map(this::fetchCreditScore), and under load unrelated endpoints that use CompletableFuture.supplyAsync become slow. Explain the link and propose a fix, with numbers for 2,000 customers, a 150 ms remote call and a 16-core server.

Key takeaways

Check yourself

References