Java Stream Gatherers
Last modified: June 8, 2025
The Gatherer interface was introduced in Java 22 as a preview feature
and became stable in later versions. It provides a powerful way to create custom
stream operations that can maintain state and perform complex transformations.
Gatherers bridge the gap between simple mapping operations and full collectors,
offering more flexibility for intermediate stream operations.
A Gatherer is a specialized form of collector that can transform
a stream of input elements into a stream of output elements. Unlike traditional
collectors that reduce streams to single values, gatherers can produce zero,
one, or many output elements for each input element while maintaining state
throughout the processing.
Gatherers excel at operations like windowing, stateful filtering, running calculations, and custom grouping scenarios where traditional stream operations fall short. They provide imperative-style control within functional pipelines while maintaining stream processing efficiency.
Gatherer Basics
A Gatherer consists of four main components:
- Initializer: Creates the initial state
- Integrator: Processes each input element with the current state
- Combiner: Combines states in parallel processing (optional)
- Finisher: Handles final state cleanup and remaining output (optional)
The method signature for creating a sequential gatherer is:
public static <T, A, R> Gatherer<T, A, R> ofSequential(
Supplier<A> initializer,
Gatherer.Integrator<A, T, R> integrator
)
where T is the input type, A is the state type, and
R is the output type.
Simple Sliding Window
This example demonstrates a basic sliding window gatherer that groups consecutive elements into overlapping windows of a specified size.
import java.util.*;
import java.util.stream.Gatherer;
void main() {
List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Gatherer<Integer, ArrayList<Integer>, List<Integer>> slidingWindow =
Gatherer.<Integer, ArrayList<Integer>, List<Integer>>ofSequential(
ArrayList::new,
(buffer, element, downstream) -> {
buffer.add(element);
if (buffer.size() == 3) {
List<Integer> window = new ArrayList<>(buffer);
buffer.remove(0); // Slide the window
return downstream.push(window);
}
return true;
}
);
List<List<Integer>> windows = numbers.stream()
.gather(slidingWindow)
.toList();
System.out.println("Original: " + numbers);
System.out.println("Sliding windows: " + windows);
}
This creates overlapping windows of size 3, demonstrating how gatherers can maintain state (the buffer) and produce multiple outputs. The integrator adds each element to the buffer and emits a window when it reaches the target size.
Running Statistics
Gatherers excel at maintaining running calculations. This example calculates running averages while processing a stream of numbers.
import java.util.stream.Gatherer;
record RunningStats(double sum, int count, double average) {}
void main() {
List<Double> prices = List.of(10.5, 25.75, 8.2, 15.0, 30.25, 12.8, 22.1);
Gatherer<Double, RunningStats, RunningStats> runningAverage =
Gatherer.<Double, RunningStats, RunningStats>ofSequential(
() -> new RunningStats(0.0, 0, 0.0),
(stats, price, downstream) -> {
double newSum = stats.sum() + price;
int newCount = stats.count() + 1;
double newAverage = newSum / newCount;
RunningStats newStats = new RunningStats(newSum, newCount, newAverage);
return downstream.push(newStats);
}
);
prices.stream()
.gather(runningAverage)
.forEach(stats ->
System.out.printf("Count: %d, Average: %.2f%n",
stats.count(), stats.average()));
}
This gatherer maintains running statistics, updating the sum, count, and average with each new element. The state is encapsulated in a record, making the code clean and type-safe. Each output shows the cumulative statistics up to that point.
Stateful Filtering
Gatherers can perform complex filtering operations that depend on previously seen elements. This example filters out consecutive duplicates.
import java.util.Optional;
import java.util.stream.Gatherer;
void main() {
List<String> words = List.of("apple", "apple", "banana", "banana",
"banana", "cherry", "apple", "cherry");
Gatherer<String, Optional<String>, String> distinctConsecutive =
Gatherer.<String, Optional<String>, String>ofSequential(
Optional::empty,
(lastSeen, current, downstream) -> {
if (lastSeen.isEmpty() || !lastSeen.get().equals(current)) {
downstream.push(current);
return Optional.of(current);
}
return lastSeen; // Skip duplicate
}
);
List<String> result = words.stream()
.gather(distinctConsecutive)
.toList();
System.out.println("Original: " + words);
System.out.println("Distinct consecutive: " + result);
}
This gatherer maintains the last seen element as state and only emits elements
that differ from their predecessor. The state is updated with each new element,
and duplicates are filtered out by not calling downstream.push().
Custom Grouping with Finisher
This example demonstrates using a finisher to handle remaining state when the stream ends. It groups consecutive elements that satisfy a condition.
import java.util.stream.Gatherer;
class GroupState {
char currentLetter = '\0';
List<String> currentGroup = new ArrayList<>();
}
void main() {
List<String> words = List.of("apple", "ant", "banana", "bear", "cat", "car", "dog");
Gatherer<String, GroupState, List<String>> groupBySameLetter =
Gatherer.<String, GroupState, List<String>>ofSequential(
GroupState::new,
(state, element, downstream) -> {
char firstLetter = element.charAt(0);
// If new letter and group not empty, emit current group
if (state.currentLetter != firstLetter && !state.currentGroup.isEmpty()) {
List<String> groupToEmit = new ArrayList<>(state.currentGroup);
state.currentGroup.clear();
downstream.push(groupToEmit);
}
// Add current element to group
state.currentLetter = firstLetter;
state.currentGroup.add(element);
return true;
},
(state, downstream) -> {
// Finisher: emit the last group if not empty
if (!state.currentGroup.isEmpty()) {
downstream.push(new ArrayList<>(state.currentGroup));
}
}
);
List<List<String>> groups = words.stream()
.gather(groupBySameLetter)
.toList();
System.out.println("Words: " + words);
System.out.println("Grouped by first letter: " + groups);
}
The finisher ensures that the last group is emitted even if the stream ends while building it. This pattern is essential for gatherers that accumulate elements before emitting them, preventing data loss at stream boundaries.
Built-in Gatherers
Java provides several built-in gatherers in the Gatherers utility
class. This example demonstrates some commonly used ones.
import java.util.stream.Gatherers;
void main() {
List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
// Fixed-size windowing
System.out.println("=== Fixed Windows (size 3) ===");
numbers.stream()
.gather(Gatherers.windowFixed(3))
.forEach(System.out::println);
// Sliding windows
System.out.println("\n=== Sliding Windows (size 3) ===");
numbers.stream()
.gather(Gatherers.windowSliding(3))
.forEach(System.out::println);
// Scanning (running totals)
System.out.println("\n=== Running Totals ===");
numbers.stream()
.gather(Gatherers.scan(() -> 0, Integer::sum))
.forEach(System.out::println);
// Folding with custom operation
System.out.println("\n=== Running Max ===");
numbers.stream()
.gather(Gatherers.fold(() -> Integer.MIN_VALUE, Integer::max))
.forEach(System.out::println);
}
Built-in gatherers provide common operations like windowing, scanning, and folding. They're optimized and ready to use, but custom gatherers give you complete control over the processing logic for specialized requirements.
Batch Processing
This example shows how to create a gatherer that processes elements in batches, useful for bulk operations or API calls with size limits.
import java.util.stream.Gatherer;
void main() {
List<String> items = List.of("A", "B", "C", "D", "E", "F", "G", "H", "I", "J");
Gatherer<String, ArrayList<String>, List<String>> batcher =
Gatherer.<String, ArrayList<String>, List<String>>ofSequential(
ArrayList::new,
(batch, item, downstream) -> {
batch.add(item);
if (batch.size() == 3) {
List<String> completeBatch = new ArrayList<>(batch);
batch.clear();
return downstream.push(completeBatch);
}
return true;
},
(batch, downstream) -> {
// Emit remaining items as final batch
if (!batch.isEmpty()) {
downstream.push(new ArrayList<>(batch));
}
}
);
items.stream()
.gather(batcher)
.forEach(batch -> System.out.println("Batch: " + batch));
}
This gatherer accumulates elements into batches of size 3, emitting complete batches as they're formed. The finisher ensures that any remaining elements form a final partial batch, preventing data loss.
Conditional State Transitions
Advanced gatherers can implement state machines. This example tracks sequences and emits alerts when certain patterns are detected.
import java.util.stream.Gatherer;
enum State { NORMAL, RISING, FALLING }
record StateInfo(State current, double lastValue, int sequenceCount) {}
void main() {
List<Double> values = List.of(1.0, 2.0, 3.0, 4.0, 3.0, 2.0, 1.0, 0.5, 1.0, 2.0);
Gatherer<Double, StateInfo, String> trendDetector =
Gatherer.<Double, StateInfo, String>ofSequential(
() -> new StateInfo(State.NORMAL, 0.0, 0),
(stateInfo, value, downstream) -> {
if (stateInfo.lastValue() == 0.0) {
return new StateInfo(State.NORMAL, value, 0);
}
State newState;
int newCount = 1;
if (value > stateInfo.lastValue()) {
newState = State.RISING;
if (stateInfo.current() == State.RISING) {
newCount = stateInfo.sequenceCount() + 1;
}
} else if (value < stateInfo.lastValue()) {
newState = State.FALLING;
if (stateInfo.current() == State.FALLING) {
newCount = stateInfo.sequenceCount() + 1;
}
} else {
newState = State.NORMAL;
}
// Emit alert for long sequences
if (newCount >= 3) {
downstream.push(String.format("Alert: %s trend detected at value %.1f (sequence: %d)",
newState, value, newCount));
}
return new StateInfo(newState, value, newCount);
}
);
System.out.println("Values: " + values);
System.out.println("Trend alerts:");
values.stream()
.gather(trendDetector)
.forEach(System.out::println);
}
This sophisticated gatherer implements a state machine that tracks rising and falling trends, emitting alerts when sequences exceed a threshold. It demonstrates how gatherers can maintain complex state and perform pattern recognition across stream elements.
Performance Considerations
This example compares gatherer performance with traditional stream operations for a computationally intensive task.
import java.util.stream.Gatherer;
import java.util.stream.IntStream;
void main() {
int size = 1_000_000;
// Gatherer approach: running sum with transformation
long gathererTime = measureTime(() -> {
IntStream.range(1, size)
.boxed()
.gather(Gatherer.<Integer, int[], String>ofSequential(
() -> new int[1],
(sum, value, downstream) -> {
sum[0] += value;
if (sum[0] % 1000 == 0) {
return downstream.push("Checkpoint: " + sum[0]);
}
return true;
}
))
.count();
});
// Traditional approach: stateful with external variable
long traditionalTime = measureTime(() -> {
final int[] sum = {0};
IntStream.range(1, size)
.boxed()
.filter(value -> {
sum[0] += value;
return sum[0] % 1000 == 0;
})
.map(value -> "Checkpoint: " + sum[0])
.count();
});
System.out.println("Gatherer time: " + gathererTime + "ms");
System.out.println("Traditional time: " + traditionalTime + "ms");
}
long measureTime(Runnable operation) {
long start = System.currentTimeMillis();
operation.run();
return System.currentTimeMillis() - start;
}
Gatherers provide clean state management without external variables, often with comparable or better performance. They excel when you need multiple stateful operations in a pipeline or when state logic is complex.
Real-world Data Processing
This comprehensive example demonstrates gatherers in a business scenario, processing transaction data with multiple conditions and transformations.
import java.time.LocalDate;
import java.util.stream.Gatherer;
record Transaction(String id, double amount, LocalDate date, String category) {}
record DailyReport(LocalDate date, double totalAmount, int transactionCount, String topCategory) {}
class DailyState {
LocalDate currentDate = null;
double dailyTotal = 0.0;
int dailyCount = 0;
Map<String, Integer> categoryCount = new HashMap<>();
}
void main() {
List<Transaction> transactions = List.of(
new Transaction("T1", 100.0, LocalDate.of(2024, 1, 1), "Food"),
new Transaction("T2", 50.0, LocalDate.of(2024, 1, 1), "Transport"),
new Transaction("T3", 200.0, LocalDate.of(2024, 1, 1), "Food"),
new Transaction("T4", 75.0, LocalDate.of(2024, 1, 2), "Transport"),
new Transaction("T5", 120.0, LocalDate.of(2024, 1, 2), "Entertainment"),
new Transaction("T6", 300.0, LocalDate.of(2024, 1, 3), "Food")
);
Gatherer<Transaction, DailyState, DailyReport> dailyReportGatherer =
Gatherer.<Transaction, DailyState, DailyReport>ofSequential(
DailyState::new,
(state, transaction, downstream) -> {
// If new date and we have data, emit current day's report
if (state.currentDate != null &&
!state.currentDate.equals(transaction.date())) {
String topCategory = state.categoryCount.entrySet().stream()
.max(Map.Entry.comparingByValue())
.map(Map.Entry::getKey)
.orElse("None");
DailyReport report = new DailyReport(
state.currentDate,
state.dailyTotal,
state.dailyCount,
topCategory
);
downstream.push(report);
// Reset for new day
state.dailyTotal = 0.0;
state.dailyCount = 0;
state.categoryCount.clear();
}
// Update state with current transaction
state.currentDate = transaction.date();
state.dailyTotal += transaction.amount();
state.dailyCount++;
state.categoryCount.merge(transaction.category(), 1, Integer::sum);
return true;
},
(state, downstream) -> {
// Emit final day's report
if (state.currentDate != null) {
String topCategory = state.categoryCount.entrySet().stream()
.max(Map.Entry.comparingByValue())
.map(Map.Entry::getKey)
.orElse("None");
DailyReport report = new DailyReport(
state.currentDate,
state.dailyTotal,
state.dailyCount,
topCategory
);
downstream.push(report);
}
}
);
System.out.println("Daily Transaction Reports:");
transactions.stream()
.gather(dailyReportGatherer)
.forEach(report ->
System.out.printf("%s: $%.2f (%d transactions, top: %s)%n",
report.date(), report.totalAmount(),
report.transactionCount(), report.topCategory()));
}
This real-world example processes financial transactions to generate daily reports. The gatherer maintains complex state including running totals, counts, and category tracking. The finisher ensures the last day's report is included, demonstrating how gatherers excel at complex data processing scenarios.
Source
Java Stream Gatherer Documentation
The Gatherer interface represents a significant advancement in Java's
Stream API, providing developers with powerful tools for custom stream operations.
While more complex than simple collectors, gatherers offer unmatched flexibility
for stateful processing, complex transformations, and advanced data analysis
scenarios. They bridge the gap between functional and imperative programming
styles, allowing clean, efficient solutions to previously challenging stream
processing problems.
Author
List all Java tutorials.