Stream Gatherers
Use Stream.gather with the built-in Gatherers for batches, sliding windows, running totals and bounded concurrent calls, and write a small gatherer of your own.
stream.gather(Gatherers.windowFixed(size))
stream.gather(Gatherers.scan(() -> initial, (acc, x) -> next))
Gatherer.ofSequential(initializer, integrator)Before Java 24 the set of intermediate operations was fixed. Anything that needed to look at more than one element at a time, such as batching or a running total, meant collecting to a list first or writing a loop. Stream.gather(Gatherer), final in Java 24 (JEP 485), is a general intermediate operation: a gatherer can keep state, emit any number of elements per input, and stop the stream early.
Built-in gatherers
The java.util.stream.Gatherers class provides five:
| Gatherer | What it emits |
|---|---|
windowFixed(n) | Lists of n elements; the last may be shorter |
windowSliding(n) | Overlapping lists of n, moving one element at a time |
scan(initial, fn) | Each intermediate result: running totals |
fold(initial, fn) | Only the final result, as a one-element stream |
mapConcurrent(max, fn) | Mapped values, computed on virtual threads with at most max running at once, in input order |
ids.stream()
.gather(Gatherers.windowFixed(500))
.forEach(batch -> repository.insertAll(batch));Back-end uses
windowFixed is the natural way to batch database inserts or API calls instead of one round trip per row. windowSliding gives moving averages over metrics. scan produces running balances for an account statement. mapConcurrent is a simple way to call a slow service for each element with a concurrency limit, so you do not open a thousand connections at once, while results stay in input order.
Writing your own
A gatherer is built from an optional initializer that creates state, an integrator that receives each element and a Downstream to push results to, and an optional finisher. Gatherer.ofSequential is the easiest factory. The integrator returns true to keep going or false to stop the stream, and downstream.push returns false once nobody wants more elements, so pass that result on. Integrator.ofGreedy marks an integrator that never stops early by itself.
Mistakes people make
- Holding a huge window.
windowFixed(n)keepsnelements in memory; size batches to what the database driver handles well. - Using
mapConcurrentfor CPU-heavy work. It is meant for waiting on I/O; for computation a parallel stream or a fixed thread pool fits better. - Reaching for a gatherer when an existing operation does the job.
map,filterand collectors are still clearer for simple cases.
Example
When to use it
- Insert a large CSV import into the database in batches of 500 rows with windowFixed.
- Show a running account balance next to each transaction in a statement with scan.
- Call a pricing service for each SKU with at most ten requests in flight using mapConcurrent.
More examples
Moving average with windowSliding
Each window holds three consecutive readings and moves forward one element at a time, so five readings give three averages. This smooths noisy metrics before alerting on them.
import java.util.List;
import java.util.stream.Gatherers;
void main() {
List<Integer> latenciesMs = List.of(100, 200, 300, 400, 500);
List<Double> movingAverage = latenciesMs.stream()
.gather(Gatherers.windowSliding(3))
.map(window -> window.stream().mapToInt(Integer::intValue).average().orElse(0))
.toList();
IO.println(movingAverage);
}
// Output:
// [200.0, 300.0, 400.0]A custom distinctBy gatherer
The state is a set of keys already seen. A repeated key returns true to keep the stream going without pushing anything, and a new key pushes the element and passes on push's answer.
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Gatherer;
record User(long id, String email) {}
static <T, K> Gatherer<T, ?, T> distinctBy(Function<? super T, ? extends K> key) {
return Gatherer.ofSequential(
() -> new HashSet<K>(),
Gatherer.Integrator.<Set<K>, T, T>ofGreedy((seen, element, downstream) ->
!seen.add(key.apply(element)) || downstream.push(element)));
}
void main() {
List<User> users = List.of(
new User(1, "[email protected]"),
new User(2, "[email protected]"),
new User(3, "[email protected]"));
List<User> unique = users.stream()
.gather(distinctBy((User u) -> u.email().toLowerCase(Locale.ROOT)))
.toList();
unique.forEach(IO::println);
}
// Output:
// User[id=1, [email protected]]
// User[id=2, [email protected]]Bounded concurrent calls with mapConcurrent
At most two simulated calls run at a time on virtual threads. A-1 is the slowest, yet it still comes first, because mapConcurrent emits results in input order.
import java.util.List;
import java.util.stream.Gatherers;
static String fetchStatus(String sku) {
try {
Thread.sleep(sku.equals("A-1") ? 200 : 20);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return sku + " in stock";
}
void main() {
List<String> results = List.of("A-1", "B-2", "C-3", "D-4").stream()
.gather(Gatherers.mapConcurrent(2, sku -> fetchStatus(sku)))
.toList();
results.forEach(IO::println);
}
// Output:
// A-1 in stock
// B-2 in stock
// C-3 in stock
// D-4 in stock
Discussion