A Deep Dive into Java Stream Performance

.jpg

📥 Download source code

The Speaker

.jpg

Cay Horstmann
Author of Core Java (14 editions since 1996), Java/JavaScript/Scala for the Impatient, etc.
Java Champion
https://horstmann.com

The Sponsors

Why Use Streams?

.png

Declarative (“what, not how”)

int largest = Stream.of(points)
    .filter(Points::isShort)
    .mapToInt(Points::shortLengthSquared)
    .max()
    .orElse(Integer.MIN_VALUE);

instead of

int largest = Integer.MIN_VALUE;
for (int i = 0; i < points.length; i++) {
    Point p = points[i];
    if (Points.isShort(p)) {
        int lengthSquared = Points.shortLengthSquared(p);
        if (lengthSquared > largest) {
            largest = lengthSquared;
        }
    }
}

Can be parallelized (sometimes)
Briefly gorgeous in 2014
Gatherer API in Java 25
Are everywhere in AI generated code

Happy day performance is good:

Benchmark             Mode  Cnt     Score    Error  Units
FilterMap.withArray   avgt  100  1711.600 ± 10.471  ms/op
FilterMap.withStream  avgt  100  1798.188 ± 11.118  ms/op

Intermediate Operations Build Pipeline

Trace into stream calls with debugger
Intellij: Settings → Debugger → Stepping
- ☑ Always do smart step into
- ☐ Do not step into these classes (you want to step in)
In FilterMap demo, set breakpoint to .filter(...)
Step into (F7) to see that it returns new StatelessOp<>(...).
Look at accept method of sink.
Step out (Shift+F8), step into .map(...)
Also returns new StatelessOp<>(...). Look at accept method.
Not doing any filtering or mapping yet, just setting up the pipeline
This is the pipeline:

Terminal Operation Pushes Elements

On to max, the terminal op. Skipping a few steps, set a breakpoint in AbstractPipeline.copyInto (line 566)

Note that the first branch executes:

spliterator.forEachRemaining(wrappedSink);

The spliterator is an iterator through the array elements
Step inside forEachRemaining
Step inside action.accept: First pipeline stage checks the filter predicate
The element might not pass, so set a breakpoint at downstream.accept(u) and resume (F9)
Step inside to see the map being computed
We could step more to see that a ReducingSink accepts the element, updating the maximum
All elements are pushed through the pipeline:

A Change of Strategy

.png

Let’s add limit after the map:

Yikes, twice as slow as the equivalent loop:

FilterMapLimit.withArray   avgt  100   664.089 ± 5.617  ms/op
FilterMapLimit.withStream  avgt  100  1169.370 ± 7.015  ms/op

No longer calls forEachRemaining:

boolean cancelled;
do { } while (!(cancelled = sink.cancellationRequested()) && spliterator.tryAdvance(sink));
return cancelled;

Cancellation happens when the limit is reached

Make flame graph:

java -jar target/benchmarks.jar \
    -prof async:output=flamegraph\;libPath=/opt/async-profiler/lib/libasyncProfiler.so \
    FilterMapLimit

Note how much time is spent in Sink$ChainedReference.cancellationRequested
Compare with array
JITWatch gives a clue: Sink.cancellationRequested can't be inlined
Seemingly small change in evaluation strategy can have substantial performance impact

Memory Consumption

.png

Stream pipeline setup can be noticeable in nested streams

var words = Files.lines(Path.of(textFile))
    .flatMap(l -> Stream.of(l.split(" "))) // A stream for every line
    .toList();

Beware of unnecessary boxing

result = points.stream().map(p -> Math.hypot(p.x(), p.y())).reduce(0.0, Double::sum); // 👎
result = points.stream().mapToDouble(p -> Math.hypot(p.x(), p.y())).sum(); // 👍

Be aware of buffering: distinct, sorted must buffer elements

Parallel Streams

Parallel stream splits into subranges

int largest = Stream.of(points)
    .parallel()
    .filter(Points::isShort)
    .mapToInt(Points::shortLengthSquared)
    .max()
    .orElse(Integer.MIN_VALUE);

Subrange tasks execute concurrently (one thread per core)

Happy day scenario: n-fold speedup

Benchmark                                           Mode  Cnt        Score       Error  Units
ParallelFilterMap.withParallelStream                avgt  100   154229.126 ±  2323.488  us/op
ParallelFilterMap.withSerialStream                  avgt  100  1828765.262 ±  9945.505  us/op

Not All Days Are Happy

What is wrong with this pipeline?

long result = Stream.of(points)
    .parallel()
    .mapToLong(Points::lengthSquared)
    .filter(p -> p < DIST)
    .limit(POINTS / 10)
    .max()
    .orElse(Integer.MIN_VALUE);

Serial is faster!

ParallelMapFilterLimit.withParallelStream           avgt  100  275946.600 ± 14045.715  us/op
ParallelMapFilterLimit.withSerialStream             avgt  100  236369.864 ±  2717.238  us/op

Problem: Parallelization is limited since we don’t know how many elements are needed
Parallel map never speculatively executes—assumes that the mapper is expensive

Remedy:

long result = Stream.of(state.points)
    .parallel()
    .unordered()
    ...

ParallelMapFilterLimit.withUnorderedParallelStream  avgt  100   56743.103 ±  1841.688  us/op

Lines in a File

More and more library methods produce streams, e.g. Files.lines

Let's parallelize and count the number of threads:

Files.lines(Path.of(filename))
    .parallel()
    .collect(Collectors.groupingBy(l -> Thread.currentThread().toString(),
        Collectors.counting()))
    .forEach((k,v) -> System.out.printf("%-40s %15d%n", k, v))

With a 1.9G file: 8 seconds

ForkJoinPool.commonPool-worker-29          16262697
ForkJoinPool.commonPool-worker-13          16616234
ForkJoinPool.commonPool-worker-11          16828071
ForkJoinPool.commonPool-worker-5           17058156
ForkJoinPool.commonPool-worker-23          16262697
ForkJoinPool.commonPool-worker-15          17550581
ForkJoinPool.commonPool-worker-17          17676845
ForkJoinPool.commonPool-worker-31          13434402
ForkJoinPool.commonPool-worker-1           16262698
ForkJoinPool.commonPool-worker-25          13411912
ForkJoinPool.commonPool-worker-9           16969771
ForkJoinPool.commonPool-worker-19          16969771
ForkJoinPool.commonPool-worker-27          16262698

With a 10G file: 167 seconds

ForkJoinPool.commonPool-worker-29             58368
ForkJoinPool.commonPool-worker-13          50244608
ForkJoinPool.commonPool-worker-11             62464
ForkJoinPool.commonPool-worker-5              41984
ForkJoinPool.commonPool-worker-23         266399744
ForkJoinPool.commonPool-worker-15         140628992
ForkJoinPool.commonPool-worker-17         237836288
ForkJoinPool.commonPool-worker-31             40960
ForkJoinPool.commonPool-worker-1              60416
ForkJoinPool.commonPool-worker-25         216946688
ForkJoinPool.commonPool-worker-9              33792
ForkJoinPool.commonPool-worker-27             34816
ForkJoinPool.commonPool-worker-3          161350656

Files.lines optimized for files up to 2GB
For larger files, falls back on non-splittable implementation

Starving the Pool

This task takes one second:

String task(int i) {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException _) {
    }
    return i + " " + Thread.currentThread().toString();
}

Now in a sequential stream:

IntStream.range(0, 60).mapToObj(this::task).toList()

How long?

60 seconds

Parallel stream:

IntStream.range(0, 60).parallel().mapToObj(this::task).toList()

How long?

5 seconds (on my machine)

But it starves the global fork-join pool
Really bad if other concurrent tasks do the same

Remedy:

IntStream.range(0, 60).boxed().gather(Gatherers.mapConcurrent(60, this::task)).toList()

How long?

1 second

Interrupting Parallel Work

What happens when the limit is reached in this pipeline:

var result = IntStream.range(0, 100)
    .boxed()
    .gather(Gatherers.mapConcurrent(10, this::task))
    .limit(20)
    .toList();

Unneeded tasks that haven't yet started won't be started
Tasks that have started will be interrupted
As always with virtual thread workloads, they should then cancel themselves
What happens when one of the tasks throws an exception?
Same behavior
When the last thread joins, the exception is rethrown
That's arguably better than with parallel streams
- There, started tasks are not interrupted
- And the exception is thrown immediately
What happens when the thread running the pipeline is interrupted?
Nothing (!)—all tasks run to completion
After all, gather can't throw an InterruptedException

Gatherers

.jpg

JEP 485 introduces gather for intermediate operations
Five built-in gatherers
You already saw mapConcurrent

Fixed and sliding window:

IntStream.range(0, 10).boxed().gather(Gatherers.windowFixed(4)).toList()
    // [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9]]
IntStream.range(0, 10).boxed().gather(Gatherers.windowSliding(4)).toList()
    // [[0, 1, 2, 3], [1, 2, 3, 4], [2, 3, 4, 5], [3, 4, 5, 6], [4, 5, 6, 7], [5, 6, 7, 8], [6, 7, 8, 9]]

Useful to compare/combine/group adjacent elements

Scan and fold (not common):

Integer[] digits = { 1, 7, 2, 9 };
Stream.of(digits).gather(Gatherers.scan(() -> 0, (x, y) -> x * 10 + y)) // [1, 17, 172, 1729]
Stream.of(digits).gather(Gatherers.fold(() -> 0, (x, y) -> x * 10 + y)) // [1729]

Additional gatherers: gatherers4j, more-gatherers, packrat

Implementing Gatherers

Example: zip with index ●●● ... → (0, ●), (1, ●), (2, ●) ...

static <T> Gatherer<T, ?, Map.Entry<Long, T> zipWithIndex() {
    class Index { long index = 0; }
    return Gatherer.ofSequential(
            Index::new,
            Gatherer.Integrator.ofGreedy((state, element, downstream) ->
                    return downstream.push(Map.entry(state.index++, element))));
}

“Integrate” upstream elements into state, send results downstream when ready
Gatherer can be sequential or parallelizable
Declared with Gatherer.ofSequential/Gatherer.of
Parallelizable gatherer combines states
Not always feasible, e.g. windowSliding, zipWithIndex
Integrator can be greedy or “picky”
Declared with Gatherer.Integrator.ofGreedy/Gatherer.Integrator.of
A greedy gatherer never short-circuits
Basel One 2025: When Greed is Good

More Flatmap Oddities

A stream inside flatMap is never parallel:

var users = List.of("Alice", "Bob");
var result = users.stream()
    .flatMap(user ->
        IntStream.range(1, 100)
            .boxed()
            .parallel() // Ignored
            .map(id -> "%s → %d (%s)".formatted(user, id, Thread.currentThread().getName()))
    )
    .toList();

All mappers run on the main thread:

[Alice → 1 (main), Alice → 2 (main), ..., Bob → 99 (main)]

Cancellation does not reach inside flatMap on a parallel unordered stream:

IntStream.range(0, 10)
    .boxed()
    .parallel()
    .unordered()
    .flatMap(i -> IntStream.range(0, i).boxed().peek(IO::println))
    .limit(10)
    .toList(); // [0, 0, 1, 0, 1, 2, 0, 1, 2, 3]

More than ten elements are peeked:

0 0 0 0 1 0 0 1 2 3 1 2 3 1 0 1 2 3 1 2 1 2 0 4 0 1 2 3 4 4 5 6 7 2 3 4 5 5 6 7 3 4 5 8 6

OMG OOM

Back to reading that 10GB file:

List<String> sample = Files.lines(Path.of(largeFile))
    .parallel()
    .gather(Gatherers.windowFixed(SAMPLE_SIZE))
    .findAny();

Causes java.lang.OutOfMemoryError
Sequential gatherer forces buffering
Buffering all lines exceeds default heap size
Also a problem with limit, sorted, etc.
Remedy: unordered()

But not always:

Stream.of(0)
    .parallel()
    .unordered()
    .flatMap(_ -> Stream.generate(() -> 0))
    .skip(10)
    .findFirst(); // OOM

The End

.jpeg

Streams can lead to more maintainable code…
…when used with taste and restraint
Expect a performance penalty over hand-coded loop
Small in many common scenarios
Keep using streams for clarity
Develop a mental performance model
Understand pipeline hazards
Multiple, subtle paths to unhappiness
Don't trust the bots
Profile to find performance issues
Thank you!
🤔 🙋‍♀️ Questions?
Feedback