Java 24 Stream Gatherers Tutorial with Examples

Introduction The Streams APIClick to read: Streams API has a fixed set of intermediate operations - map, filter, flatMap, distinct, sorted, limit and a few more. For terminal operations, collectorsClick to read: collectors have always let you write your own, but there was no way to add a new intermediate operation. Common needs - grouping elements into fixed-size batches, computing a running total, removing duplicates by a key - had to be done outside the stream. Stream gatherers, previewed in Java 22 and 23 and final in Java 24Click to read tutorial: Overview of New Features in Java 24 (JEP 485), fill this gap: stream.gather(gatherer) applies any intermediate operation, and Java 24 ships a set of ready-made gatherers. This tutorial explains stream gatherers with examples - the built-in gatherers, and how to write your own. Built-in gatherers: windows, fold and scan The class java.util.stream.Gatherers provides the most commonly needed gatherers -
windowFixed, windowSliding, fold and scan
//BuiltInGatherers.java
import java.util.List;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class BuiltInGatherers {
  public static void main(String[] args) {
    // Fixed-size batches, e.g. for bulk database inserts
    List<List<Integer>> batches = Stream.of(1, 2, 3, 4, 5, 6, 7).gather(Gatherers.windowFixed(3)).toList();
    System.out.println("windowFixed(3)  : " + batches);

    // Overlapping windows, e.g. for moving averages
    List<Double> movingAverage = Stream.of(10, 20, 30, 40, 50)
        .gather(Gatherers.windowSliding(3))
        .map(w -> w.stream().mapToInt(Integer::intValue).average().orElse(0))
        .toList();
    System.out.println("3-day moving avg: " + movingAverage);

    // Running total: emits every intermediate result
    List<Integer> running = Stream.of(5, 10, 3, 7).gather(Gatherers.scan(() -> 0, Integer::sum)).toList();
    System.out.println("scan (running)  : " + running);

    // fold: like reduce, but the result can have a different type, and it stays a stream
    String joined = Stream.of("a", "b", "c")
        .gather(Gatherers.fold(() -> "", (acc, s) -> acc.isEmpty() ? s : acc + "-" + s))
        .findFirst().orElse("");
    System.out.println("fold            : " + joined);
  }
}
 OUTPUT of the above code
windowFixed(3)  : [[1, 2, 3], [4, 5, 6], [7]]
3-day moving avg: [20.0, 30.0, 40.0]
scan (running)  : [5, 15, 18, 25]
fold            : a-b-c
Explanation of the code
  • windowFixed(3) groups elements into lists of 3; the last window can be smaller. windowSliding(3) produces overlapping windows, each moving forward by one element.
  • scan(initial, function) emits the accumulated value after every element - a running total - while fold(initial, function) emits only the final value, as a one-element stream.
  • Each gatherer returns a stream, so more operations can follow it, and gatherers can be chained with andThen().
mapConcurrent: concurrent calls with virtual threads Gatherers.mapConcurrent(maxConcurrency, function) applies a function to elements using virtual threadsClick to read: virtual threads, running up to maxConcurrency of them at a time, while keeping the results in the original order. It is ideal for slow I/O calls, such as fetching a list of URLs -
Concurrent, ordered I/O calls with mapConcurrent
//MapConcurrentDemo.java
import java.util.List;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class MapConcurrentDemo {
  static String slowLookup(int id) {
    try { Thread.sleep(500); } catch (InterruptedException e) { throw new RuntimeException(e); }  // a remote call
    return "user-" + id;
  }

  public static void main(String[] args) {
    long start = System.nanoTime();
    List<String> users = Stream.of(1, 2, 3, 4, 5, 6, 7, 8)
        .gather(Gatherers.mapConcurrent(4, MapConcurrentDemo::slowLookup))
        .toList();
    long seconds = Math.round((System.nanoTime() - start) / 1e9);
    System.out.println(users);
    System.out.println("8 calls of 0.5 s with 4 at a time took about " + seconds + " s");
  }
}
 OUTPUT of the above code
[user-1, user-2, user-3, user-4, user-5, user-6, user-7, user-8]
8 calls of 0.5 s with 4 at a time took about 1 s
Eight half-second calls, four at a time, take about one second instead of four, and the results come back in order. Writing your own gatherer A Gatherer<T, A, R> takes elements of type T, keeps optional state of type A, and pushes elements of type R downstream. It is made of up to four functions -
  • an initializer, which creates the private state,
  • an integrator, called for each element, which may push zero or more results downstream and returns false to stop the stream early,
  • an optional finisher, called at the end, which can push remaining results, and
  • an optional combiner, for parallel streams.
The example below writes a gatherer which removes duplicates by a key - something distinct() cannot do - and one which stops the stream once a total is reached -
distinctBy and takeUntilTotal gatherers
//CustomGatherers.java
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class CustomGatherers {
  // Keep the first element for each key
  static <T, K> Gatherer<T, Set<K>, T> distinctBy(Function<T, K> key) {
    return Gatherer.ofSequential(
        HashSet::new,                                                  // initializer: the keys seen
        (seen, element, downstream) ->                                 // integrator
            !seen.add(key.apply(element)) || downstream.push(element));
  }

  // Pass elements until their running total reaches a limit, then stop the stream
  static Gatherer<Integer, int[], Integer> takeUntilTotal(int limit) {
    return Gatherer.ofSequential(
        () -> new int[1],
        (total, amount, downstream) -> {
          total[0] += amount;
          return downstream.push(amount) && total[0] < limit;          // false = stop
        });
  }

  public static void main(String[] args) {
    record Person(String name, String city) { }
    List<Person> people = List.of(new Person("Tom", "Pune"), new Person("Harry", "Delhi"),
        new Person("Sam", "Pune"), new Person("Ann", "Delhi"), new Person("Raj", "Goa"));
    System.out.println("First person per city: "
        + people.stream().gather(distinctBy(Person::city)).map(Person::name).toList());

    System.out.println("Orders until 100: "
        + Stream.of(30, 25, 40, 50, 10).gather(takeUntilTotal(100)).toList());
  }
}
 OUTPUT of the above code
First person per city: [Tom, Harry, Raj]
Orders until 100: [30, 25, 40, 50]
Gatherer.ofSequential() creates a gatherer which runs sequentially, even in a parallel stream; Gatherer.of() creates one which can run in parallel if a combiner is given. downstream.push() returns false when downstream needs no more elements, so a well-behaved integrator passes that result on, as both examples do.
Gatherers vs collectors A collector consumes the whole stream and produces one final result - it ends the pipeline. A gatherer transforms the stream in the middle of the pipeline, can emit any number of elements at any time, can stop early, and can be followed by more operations. Together, gather() and collect() make the Streams API fully extensible at both ends. Summary In this tutorial we saw how stream gatherers, final in Java 24, let you add your own intermediate operations to streams with gather(). We used the built-in windowFixed, windowSliding, scan and fold gatherers, ran slow calls concurrently and in order with mapConcurrent, and wrote custom gatherers with an initializer and an integrator, including one which stops the stream early. Next, read about the Class-File APIClick to read tutorial: Java 24 Class-File API Tutorial with Examples.