Gatherers, Parallel Streams, and Stream Cost

This lesson extends Streams and Collectors with custom intermediate operations, the Spliterator that feeds every stream, parallel execution, and what a stream costs compared with a loop. After it, a reader can write a gatherer, decide when a parallel stream helps, and explain why hot paths often use loops.

Idea or API	Release
`Spliterator`, `parallelStream`, `parallel()`	Java 8
`Stream.gather`, `Gatherer`, `Gatherers`	Java 24 (preview in 22 and 23, JEP 485)

Example 1: built-in gatherers

A gatherer is an intermediate operation passed to Stream.gather. Unlike map or filter, it can keep state, emit more or fewer elements than it receives, and stop the stream early. The Gatherers class provides common ones.

Java

import java.util.List;
import java.util.stream.Gatherers;

public class BuiltInGatherers {
    public static void main(String[] args) {
        List<Integer> prices = List.of(1, 2, 3, 4, 5, 6, 7);

        System.out.println(prices.stream().gather(Gatherers.windowFixed(3)).toList());
        System.out.println(prices.stream().limit(4).gather(Gatherers.windowSliding(2)).toList());
        System.out.println(prices.stream().limit(4)
                .gather(Gatherers.scan(() -> 0, Integer::sum)).toList());
        System.out.println(prices.stream()
                .gather(Gatherers.fold(() -> 0, Integer::sum)).toList());

        List<Double> movingAverage = prices.stream()
                .gather(Gatherers.windowSliding(3))
                .map(w -> w.stream().mapToInt(Integer::intValue).average().orElseThrow())
                .toList();
        System.out.println(movingAverage);
    }
}

Output

[[1, 2, 3], [4, 5, 6], [7]]
[[1, 2], [2, 3], [3, 4]]
[1, 3, 6, 10]
[28]
[2.0, 3.0, 4.0, 5.0, 6.0]

Gatherer	Output
`windowFixed(n)`	consecutive lists of `n`; the last one may be shorter
`windowSliding(n)`	overlapping lists of `n`, moving by one element
`scan(init, f)`	one running result per element
`fold(init, f)`	one final result, emitted when the input ends
`mapConcurrent(n, f)`	applies `f` to up to `n` elements at once, keeps encounter order

Before gatherers, a sliding window over a stream needed a custom collector or an index-based loop. fold differs from reduce in that its result is a stream element, so the pipeline can continue after it.

Example 2: `mapConcurrent` keeps the order

mapConcurrent runs the mapping function for several elements at the same time on virtual threads, which are covered in Virtual Threads in Depth. The output stays in encounter order even when later elements finish first. It fits blocking calls such as remote lookups.

Java

import java.util.List;
import java.util.stream.Gatherers;

public class MapConcurrentOrder {
    static String slowLookup(int id) {
        try {
            Thread.sleep(50 - id * 10);   // later ids finish first
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "price-" + id;
    }

    public static void main(String[] args) {
        List<String> prices = List.of(1, 2, 3, 4).stream()
                .gather(Gatherers.mapConcurrent(4, MapConcurrentOrder::slowLookup))
                .toList();
        System.out.println(prices);
    }
}

Output

[price-1, price-2, price-3, price-4]

Element 4 sleeps the shortest time, but the result list still starts with price-1. The first argument caps how many calls run at once, which protects the remote service. If a call throws, the exception is rethrown from the stream as a RuntimeException and the remaining calls are cancelled (JDK source, Gatherers.mapConcurrent).

Example 3: a custom sequential gatherer

Gatherer.ofSequential builds a gatherer from an initializer, which creates the state, and an integrator, which handles one element and may push results to the downstream (the next stage). The integrator returns false to stop the stream.

Java

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Gatherer;

public class CustomGatherer {
    record Tick(String symbol, long price) {}

    static <T, K> Gatherer<T, ?, T> distinctBy(Function<T, K> key) {
        return Gatherer.<T, Set<K>, T>ofSequential(
                HashSet::new,
                (seen, element, downstream) ->
                        !seen.add(key.apply(element)) || downstream.push(element));
    }

    static <T> Gatherer<T, ?, T> limitTo(int maxCount) {
        return Gatherer.<T, int[], T>ofSequential(
                () -> new int[1],
                (count, element, downstream) -> {
                    if (count[0]++ >= maxCount) return false;   // stop the stream
                    return downstream.push(element);
                });
    }

    public static void main(String[] args) {
        List<Tick> ticks = List.of(
                new Tick("AAPL", 190), new Tick("MSFT", 410),
                new Tick("AAPL", 191), new Tick("IBM", 90), new Tick("MSFT", 411));

        System.out.println(ticks.stream().gather(distinctBy(Tick::symbol))
                .map(Tick::symbol).toList());
        System.out.println(ticks.stream().gather(limitTo(2))
                .map(Tick::price).toList());
    }
}

Output

[AAPL, MSFT, IBM]
[190, 410]

downstream.push returns false when the downstream wants no more elements, and the integrator passes that on. For a key already seen, !seen.add(...) is true, so the element is dropped and the stream continues. A new state is created for every stream that uses the gatherer, so the same gatherer object can be reused.

Example 4: a gatherer with a finisher

A finisher runs once after the last element and can push the remaining state. This gatherer groups ticks into batches whose total quantity reaches a limit, a "volume bar" in market-data terms. Without the finisher the last, partial batch would be lost.

Java

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Gatherer;

public class VolumeBars {
    static Gatherer<Integer, ?, List<Integer>> volumeBars(int limit) {
        class State {
            List<Integer> batch = new ArrayList<>();
            int volume;
        }
        return Gatherer.<Integer, State, List<Integer>>ofSequential(
                State::new,
                (state, qty, downstream) -> {
                    state.batch.add(qty);
                    state.volume += qty;
                    if (state.volume < limit) return true;
                    boolean more = downstream.push(List.copyOf(state.batch));
                    state.batch.clear();
                    state.volume = 0;
                    return more;
                },
                (state, downstream) -> {
                    if (!state.batch.isEmpty()) downstream.push(List.copyOf(state.batch));
                });
    }

    public static void main(String[] args) {
        List<Integer> quantities = List.of(40, 30, 50, 100, 10, 20);
        System.out.println(quantities.stream().gather(volumeBars(100)).toList());
        System.out.println(quantities.stream().gather(volumeBars(100)).limit(1).toList());
    }
}

Output

[[40, 30, 50], [100], [10, 20]]
[[40, 30, 50]]

The local class State holds mutable fields, which a lambda cannot assign directly when they are local variables. List.copyOf hands the downstream an unmodifiable snapshot, so clearing the batch does not change emitted lists. With limit(1) the push returns false, the integrator returns false, and no more elements are read.

Example 5: `Spliterator` and characteristics

A Spliterator is the source object behind every stream. It walks elements (tryAdvance), can split off part of its elements for another worker (trySplit), and reports characteristics, bit flags that describe the source. The stream uses them to skip work: a SORTED source lets sorted() do nothing, and a SIZED source lets toList allocate the exact array size.

Java

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Spliterator;
import java.util.TreeSet;
import java.util.stream.IntStream;

public class SpliteratorCharacteristics {
    static String flags(Spliterator<?> s) {
        List<String> names = new ArrayList<>();
        if (s.hasCharacteristics(Spliterator.ORDERED)) names.add("ORDERED");
        if (s.hasCharacteristics(Spliterator.DISTINCT)) names.add("DISTINCT");
        if (s.hasCharacteristics(Spliterator.SORTED)) names.add("SORTED");
        if (s.hasCharacteristics(Spliterator.SIZED)) names.add("SIZED");
        if (s.hasCharacteristics(Spliterator.SUBSIZED)) names.add("SUBSIZED");
        if (s.hasCharacteristics(Spliterator.NONNULL)) names.add("NONNULL");
        if (s.hasCharacteristics(Spliterator.IMMUTABLE)) names.add("IMMUTABLE");
        return names.toString();
    }

    public static void main(String[] args) {
        List<Integer> arrayList = new ArrayList<>(List.of(1, 2, 3, 4, 5, 6, 7, 8));
        System.out.println("ArrayList   " + flags(arrayList.spliterator()));
        System.out.println("HashSet     " + flags(new HashSet<>(arrayList).spliterator()));
        System.out.println("TreeSet     " + flags(new TreeSet<>(arrayList).spliterator()));
        System.out.println("IntStream   " + flags(IntStream.range(0, 8).spliterator()));
        System.out.println("filtered    " + flags(arrayList.stream().filter(i -> i > 2).spliterator()));

        Spliterator<Integer> right = arrayList.spliterator();
        Spliterator<Integer> left = right.trySplit();
        System.out.println("split sizes " + left.estimateSize() + " + " + right.estimateSize());
        left.tryAdvance(i -> System.out.println("first of left = " + i));
    }
}

Output

ArrayList   [ORDERED, SIZED, SUBSIZED]
HashSet     [DISTINCT, SIZED]
TreeSet     [ORDERED, DISTINCT, SORTED, SIZED]
IntStream   [ORDERED, DISTINCT, SORTED, SIZED, SUBSIZED, NONNULL, IMMUTABLE]
filtered    [ORDERED]
split sizes 4 + 4
first of left = 1

ArrayList splits in half by index, which is cheap and balanced, so array-backed sources split well for parallel work. After filter, SIZED is gone, because the number of matches is unknown until the filter runs. HashSet has no ORDERED flag, which matches the unspecified iteration order of a HashMap.

Example 6: parallel streams and their rules

parallelStream() or .parallel() splits the source with trySplit and runs the parts on the common fork-join pool, a shared thread pool whose size defaults to the number of CPUs minus one; it is covered in Executors and CompletableFuture. Results are correct only when the operations are stateless, free of side effects, and the reduction is associative with a true identity.

Java

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;

public class ParallelRules {
    public static void main(String[] args) {
        System.out.println("common pool parallelism > 0: "
                + (ForkJoinPool.commonPool().getParallelism() > 0));

        long sequential = LongStream.rangeClosed(1, 1_000_000).sum();
        long parallel = LongStream.rangeClosed(1, 1_000_000).parallel().sum();
        System.out.println("sum " + parallel + ", equal = " + (sequential == parallel));

        int good = IntStream.rangeClosed(1, 100).parallel().reduce(0, Integer::sum);
        int bad = IntStream.rangeClosed(1, 100).parallel().reduce(10, Integer::sum);
        System.out.println("identity 0: " + good + ", identity 10 matches sequential 5060: " + (bad == 5060));

        int nonAssociative = IntStream.rangeClosed(1, 100).parallel().reduce(0, (a, b) -> a - b);
        int sequentialMinus = IntStream.rangeClosed(1, 100).reduce(0, (a, b) -> a - b);
        System.out.println("subtraction sequential " + sequentialMinus
                + ", parallel matches: " + (nonAssociative == sequentialMinus));

        List<Integer> ordered = IntStream.range(0, 20).parallel().boxed()
                .map(i -> i * i).collect(Collectors.toList());
        System.out.println("collect keeps order: " + ordered.equals(
                IntStream.range(0, 20).map(i -> i * i).boxed().toList()));
        System.out.println("findFirst = " + IntStream.range(0, 1000).parallel()
                .filter(i -> i % 100 == 99).findFirst().orElseThrow());
    }
}

Output

common pool parallelism > 0: true
sum 500000500000, equal = true
identity 0: 5050, identity 10 matches sequential 5060: false
subtraction sequential -5050, parallel matches: false
collect keeps order: true
findFirst = 99

The identity 10 is added once per split part, not once in total, so the result is larger than 5,060, which is the sequential answer. Subtraction is not associative, so the parallel result differs from the sequential one. collect and findFirst respect encounter order even in parallel; forEach and findAny do not, and forEachOrdered restores order at a cost. A parallel stream helps when the source splits well (arrays, ArrayList, IntStream.range), the per-element work is large, and there are many elements. It does not help for small inputs, LinkedList or Stream.iterate sources, blocking work, or code that already runs on many threads, because all parallel streams in the JVM share the one common pool.

Example 7: stream cost compared with a loop

A stream allocates a pipeline object per stage and calls each lambda through an interface call, and a Stream<Integer> boxes every value. A boxed value is an Integer object that wraps an int; Integer.valueOf caches -128 to 127 and creates a new object for other values. The three versions below compute the same result with different costs.

Java

import java.util.List;
import java.util.stream.IntStream;
import java.util.stream.Stream;

public class StreamVsLoop {
    public static void main(String[] args) {
        int[] qty = IntStream.rangeClosed(1, 10_000).toArray();

        long loop = 0;
        for (int q : qty) {
            if (q % 2 == 0) loop += q;
        }

        long primitive = IntStream.of(qty).filter(q -> q % 2 == 0).asLongStream().sum();

        long boxed = Stream.iterate(1, i -> i <= 10_000, i -> i + 1)
                .filter(q -> q % 2 == 0)
                .mapToLong(Integer::longValue)
                .sum();

        System.out.println(loop + " " + primitive + " " + boxed);
        System.out.println("small values cached: " + (Integer.valueOf(100) == Integer.valueOf(100)));
        System.out.println("large values cached: " + (Integer.valueOf(1000) == Integer.valueOf(1000)));

        List<Integer> boxes = Stream.iterate(1, i -> i + 1).limit(5_000).toList();
        long distinctObjects = boxes.stream().filter(i -> i > 127).count();
        System.out.println("values above 127, one Integer object each: " + distinctObjects);
    }
}

Output

25005000 25005000 25005000
small values cached: true
large values cached: false
values above 127, one Integer object each: 4873

All three sums match. The loop has no allocation and no indirect calls. The IntStream version avoids boxing but still builds pipeline objects and calls lambdas through interfaces; the JIT compiler often inlines these in a warmed-up method, but that is not guaranteed. The boxed version creates one Integer per value above 127, and each becomes garbage. On a hot path (code run millions of times per second, such as an order handler), these costs add allocation and GC pressure; measuring them is covered in Allocation-Free and Cache-Friendly Code. Streams fit setup code, reports, and batch work, where readability matters more than nanoseconds.

Example 8: a market-data pipeline

The final program combines a custom gatherer, built-in gatherers, and a collector: it drops out-of-order ticks per symbol, computes a sliding average, and reports the running maximum.

Java

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
import java.util.stream.Gatherer;
import java.util.stream.Gatherers;

public class MarketDataPipeline {
    record Tick(String symbol, long seq, long priceCents) {}

    static Gatherer<Tick, ?, Tick> dropStale() {
        return Gatherer.<Tick, Map<String, Long>, Tick>ofSequential(
                HashMap::new,
                (lastSeq, tick, downstream) -> {
                    Long last = lastSeq.get(tick.symbol());
                    if (last != null && tick.seq() <= last) return true;   // stale or duplicate
                    lastSeq.put(tick.symbol(), tick.seq());
                    return downstream.push(tick);
                });
    }

    public static void main(String[] args) {
        List<Tick> feed = List.of(
                new Tick("AAPL", 1, 19_000), new Tick("AAPL", 2, 19_010),
                new Tick("MSFT", 1, 41_000), new Tick("AAPL", 2, 19_010),
                new Tick("AAPL", 3, 19_040), new Tick("AAPL", 1, 18_000),
                new Tick("MSFT", 2, 41_030), new Tick("AAPL", 4, 19_030));

        List<Tick> clean = feed.stream().gather(dropStale()).toList();
        System.out.println("kept " + clean.size() + " of " + feed.size());

        Map<String, List<Long>> bySymbol = clean.stream().collect(Collectors.groupingBy(
                Tick::symbol, TreeMap::new, Collectors.mapping(Tick::priceCents, Collectors.toList())));

        bySymbol.forEach((symbol, prices) -> {
            List<Long> avg2 = prices.stream()
                    .gather(Gatherers.windowSliding(2))
                    .map(w -> (w.get(0) + w.get(1)) / 2)
                    .toList();
            List<Long> runningMax = prices.stream()
                    .gather(Gatherers.scan(() -> Long.MIN_VALUE, Math::max))
                    .toList();
            System.out.println(symbol + " prices " + prices + " avg2 " + avg2 + " max " + runningMax);
        });
    }
}

Output

kept 6 of 8
AAPL prices [19000, 19010, 19040, 19030] avg2 [19005, 19025, 19035] max [19000, 19010, 19040, 19040]
MSFT prices [41000, 41030] avg2 [41015] max [41000, 41030]

The duplicate AAPL 2 and the late AAPL 1 are dropped. windowSliding(2) on a list of two MSFT prices gives one window. A gatherer built with ofSequential keeps the stream sequential at that stage, which a stateful step like dropStale requires.

Interview points

A gatherer is a user-defined intermediate operation: initializer, integrator, optional combiner, optional finisher.
The integrator returns false to stop the stream; it passes on the result of downstream.push.
fold emits one result at the end; scan emits a running result per element.
mapConcurrent bounds concurrency, uses virtual threads, and preserves encounter order.
A Spliterator traverses and splits a source; characteristics such as SIZED, SORTED, and ORDERED let the stream skip work.
Parallel streams run on the shared common fork-join pool; a slow or blocking task there slows every parallel stream in the JVM.
A parallel reduce needs an associative function and a true identity, or results change with the split count.
Parallel helps for large, well-splitting sources with CPU-heavy per-element work; it hurts for small inputs and linked or iterate-based sources.
Streams cost pipeline objects, interface calls, and boxing for Stream<Integer>; hot paths use loops over primitive arrays.

Summary

Stream.gather and Gatherers are final in Java 24.
windowFixed, windowSliding, fold, scan, and mapConcurrent cover common stateful steps.
Gatherer.ofSequential(initializer, integrator, finisher) builds a custom stateful operation.
ArrayList and IntStream.range sources are SIZED and SUBSIZED and split evenly; filter removes SIZED.
ForkJoinPool.commonPool() runs all parallel streams.
collect and findFirst keep encounter order in parallel; forEach and findAny do not.
Integer.valueOf caches -128 to 127; other boxed values are new objects.

Example 1: built-in gatherers

Example 2: mapConcurrent keeps the order

Example 3: a custom sequential gatherer

Example 4: a gatherer with a finisher

Example 5: Spliterator and characteristics

Example 6: parallel streams and their rules

Example 7: stream cost compared with a loop

Example 8: a market-data pipeline

Interview points

Summary

Example 2: `mapConcurrent` keeps the order

Example 5: `Spliterator` and characteristics