Advanced Java Stream API Patterns

Most Java developers know filter, map, and collect. But the Stream API offers far more powerful patterns for data transformation and aggregation. This article covers advanced techniques that go beyond the basics.

Custom Collectors

The built-in Collectors utility handles most use cases, but custom collectors let you express complex aggregations in a composable way.

A Collector has four parts: a supplier (creates the mutable container), an accumulator (adds an element), a combiner (merges two containers for parallel streams), and a finisher (transforms the container into the result).

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

// Custom collector: running statistics
public class StatsCollector {

    record Stats(double min, double max, double mean, long count) {}

    static Collector<Double, double[], Stats> toStats() {
        return Collector.of(
            () -> new double[]{Double.MAX_VALUE, Double.MIN_VALUE, 0.0, 0},
            (acc, val) -> {
                acc[0] = Math.min(acc[0], val);
                acc[1] = Math.max(acc[1], val);
                acc[2] += val;
                acc[3]++;
            },
            (a, b) -> {
                a[0] = Math.min(a[0], b[0]);
                a[1] = Math.max(a[1], b[1]);
                a[2] += b[2];
                a[3] += b[3];
                return a;
            },
            acc -> new Stats(acc[0], acc[1], acc[2] / acc[3], (long) acc[3])
        );
    }

    public static void main(String[] args) {
        var stats = List.of(3.2, 1.5, 7.8, 4.1, 9.0, 2.3)
            .stream()
            .collect(toStats());

        System.out.printf("Min: %.1f, Max: %.1f, Mean: %.2f, Count: %d%n",
            stats.min(), stats.max(), stats.mean(), stats.count());
    }
}

Collecting into Immutable Structures

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

// Collector that produces an unmodifiable map with duplicate key handling
static <T, K, V> Collector<T, ?, Map<K, V>> toUnmodifiableMapMerging(
        java.util.function.Function<T, K> keyMapper,
        java.util.function.Function<T, V> valueMapper,
        java.util.function.BinaryOperator<V> mergeFunction) {

    return Collectors.collectingAndThen(
        Collectors.toMap(keyMapper, valueMapper, mergeFunction),
        Collections::unmodifiableMap
    );
}

Partitioning with Multiple Predicates

The standard partitioningBy only splits by one predicate. For multiple groups, combine groupingBy with a classifier:

record Transaction(String type, double amount, String status) {}

public static void main(String[] args) {
    var transactions = List.of(
        new Transaction("sale", 100.0, "completed"),
        new Transaction("refund", 25.0, "completed"),
        new Transaction("sale", 200.0, "pending"),
        new Transaction("sale", 50.0, "completed"),
        new Transaction("refund", 30.0, "pending")
    );

    // Multi-level grouping: type -> status -> total amount
    Map<String, Map<String, Double>> grouped = transactions.stream()
        .collect(Collectors.groupingBy(
            Transaction::type,
            Collectors.groupingBy(
                Transaction::status,
                Collectors.summingDouble(Transaction::amount)
            )
        ));

    grouped.forEach((type, statusMap) -> {
        System.out.println(type + ":");
        statusMap.forEach((status, total) ->
            System.out.printf("  %s: $%.2f%n", status, total));
    });
}

Sliding Windows and Batching

Streams do not have a built-in windowing operation, but you can build one:

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

static <T> Stream<List<T>> sliding(List<T> list, int windowSize) {
    if (list.size() < windowSize) return Stream.of(list);
    return IntStream.rangeClosed(0, list.size() - windowSize)
        .mapToObj(i -> list.subList(i, i + windowSize));
}

static <T> Stream<List<T>> batches(List<T> list, int batchSize) {
    return IntStream.iterate(0, i -> i < list.size(), i -> i + batchSize)
        .mapToObj(i -> list.subList(i, Math.min(i + batchSize, list.size())));
}

public static void main(String[] args) {
    var prices = List.of(10.0, 12.0, 11.5, 13.0, 14.5, 13.5, 15.0);

    // 3-period moving average
    System.out.println("Moving averages:");
    sliding(prices, 3)
        .map(window -> window.stream()
            .mapToDouble(d -> d)
            .average()
            .orElse(0))
        .forEach(avg -> System.out.printf("  %.2f%n", avg));

    // Process in batches of 3
    System.out.println("Batches:");
    batches(prices, 3).forEach(batch -> System.out.println("  " + batch));
}

FlatMap Patterns

flatMap is essential when each input element maps to zero or more output elements. Here are patterns beyond the basic case:

record Order(String id, List<LineItem> items) {}
record LineItem(String product, int quantity, double price) {}

public static void main(String[] args) {
    var orders = List.of(
        new Order("A", List.of(
            new LineItem("Widget", 2, 10.0),
            new LineItem("Gadget", 1, 25.0))),
        new Order("B", List.of(
            new LineItem("Widget", 5, 10.0),
            new LineItem("Doohickey", 3, 7.5)))
    );

    // Flatten and aggregate: total revenue per product
    Map<String, Double> revenueByProduct = orders.stream()
        .flatMap(order -> order.items().stream())
        .collect(Collectors.groupingBy(
            LineItem::product,
            Collectors.summingDouble(li -> li.quantity() * li.price())
        ));

    revenueByProduct.forEach((product, revenue) ->
        System.out.printf("%s: $%.2f%n", product, revenue));

    // FlatMap with Optional: filter and unwrap in one step
    List<String> names = List.of("Alice", "", "Bob", "", "Charlie");
    List<String> nonEmpty = names.stream()
        .map(s -> s.isEmpty() ? Optional.<String>empty() : Optional.of(s))
        .flatMap(Optional::stream)
        .toList();
    System.out.println(nonEmpty); // [Alice, Bob, Charlie]
}

Parallel Streams

Parallel streams split work across the common ForkJoinPool. They can speed up CPU-bound operations on large datasets but introduce pitfalls.

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

public class ParallelStreamDemo {
    public static void main(String[] args) {
        var numbers = IntStream.rangeClosed(1, 10_000_000)
            .boxed()
            .toList();

        // CPU-bound: parallel can help
        long start = System.nanoTime();
        double sum = numbers.parallelStream()
            .mapToDouble(n -> Math.sqrt(n) * Math.log(n))
            .sum();
        long elapsed = System.nanoTime() - start;
        System.out.printf("Parallel sum: %.2f in %d ms%n", sum, elapsed / 1_000_000);

        // Sequential comparison
        start = System.nanoTime();
        sum = numbers.stream()
            .mapToDouble(n -> Math.sqrt(n) * Math.log(n))
            .sum();
        elapsed = System.nanoTime() - start;
        System.out.printf("Sequential sum: %.2f in %d ms%n", sum, elapsed / 1_000_000);
    }
}

Parallel Stream Safety Rules

1. Never use parallel streams with shared mutable state:

// DANGEROUS: race condition
var results = new ArrayList<Integer>();
IntStream.range(0, 1000).parallel().forEach(results::add); // Corrupted list

// SAFE: let the stream collect
var results2 = IntStream.range(0, 1000).parallel().boxed().toList();

2. Avoid order-dependent operations in parallel:

// forEachOrdered preserves encounter order but kills parallelism benefits
stream.parallel().forEachOrdered(System.out::println); // Defeats the purpose

3. Use a custom ForkJoinPool for I/O-heavy parallel work:

import java.util.concurrent.ForkJoinPool;

var customPool = new ForkJoinPool(16);
var result = customPool.submit(() ->
    urls.parallelStream()
        .map(ParallelStreamDemo::fetchUrl)
        .toList()
).get();
customPool.shutdown();

Reduce and Fold Patterns

reduce is the most general terminal operation. Understanding its three-argument form is essential for parallel correctness:

record Sale(String region, double amount) {}

public static void main(String[] args) {
    var sales = List.of(
        new Sale("East", 100), new Sale("West", 200),
        new Sale("East", 150), new Sale("West", 300)
    );

    // Three-argument reduce for type-changing reductions
    Map<String, Double> totals = sales.stream()
        .reduce(
            new HashMap<>(),                         // Identity
            (map, sale) -> {                         // Accumulator
                var copy = new HashMap<>(map);
                copy.merge(sale.region(), sale.amount(), Double::sum);
                return copy;
            },
            (map1, map2) -> {                        // Combiner (for parallel)
                var merged = new HashMap<>(map1);
                map2.forEach((k, v) -> merged.merge(k, v, Double::sum));
                return merged;
            }
        );

    System.out.println(totals); // {East=250.0, West=500.0}
}

In practice, Collectors.groupingBy with a downstream collector is cleaner for this use case. Reserve the three-argument reduce for cases where no built-in collector fits.

Teeing Collector

The teeing collector applies two collectors simultaneously and merges their results:

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

record Range(double min, double max) {}

public static void main(String[] args) {
    var numbers = List.of(3.0, 1.0, 4.0, 1.0, 5.0, 9.0, 2.0, 6.0);

    Range range = numbers.stream()
        .collect(Collectors.teeing(
            Collectors.minBy(Double::compareTo),
            Collectors.maxBy(Double::compareTo),
            (min, max) -> new Range(
                min.orElse(0.0),
                max.orElse(0.0)
            )
        ));

    System.out.printf("Range: [%.1f, %.1f]%n", range.min(), range.max());
}

Lazy Evaluation and Short-Circuiting

Streams are lazy: intermediate operations build a pipeline that only executes when a terminal operation is called. Short-circuiting operations like findFirst, anyMatch, and limit stop processing early.

// Only processes elements until a match is found
Optional<String> first = Stream.of("apple", "banana", "cherry", "date")
    .filter(s -> {
        System.out.println("Checking: " + s);
        return s.startsWith("c");
    })
    .findFirst();
// Output: Checking: apple, Checking: banana, Checking: cherry
// "date" is never checked

Use takeWhile and dropWhile for ordered streams where you want prefix-based filtering:

var sorted = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

var lessThanFive = sorted.stream().takeWhile(n -> n < 5).toList();
System.out.println(lessThanFive); // [1, 2, 3, 4]

var fiveAndAbove = sorted.stream().dropWhile(n -> n < 5).toList();
System.out.println(fiveAndAbove); // [5, 6, 7, 8, 9, 10]

Performance Guidelines

• Prefer primitive streams (IntStream, LongStream, DoubleStream) to avoid boxing overhead
• Avoid streams for simple loops over small collections where a for-loop is clearer and faster
• Use toList() instead of collect(Collectors.toList()) in Java 16+ for readability and potential optimization
• Do not nest streams deeply; extract helper methods for clarity
• Benchmark before parallelizing; the overhead of splitting and merging often exceeds the benefit for small datasets

Wrapping Up

The Stream API is more than a syntax convenience for loops. Custom collectors let you express complex aggregations as reusable components. Parallel streams unlock multi-core processing for CPU-bound work. Patterns like sliding windows, teeing, and flatMap with Optional extend the API’s reach to real-world data processing tasks. The key is understanding when each pattern fits and when a simpler approach serves you better.