Stream Gatherers in Java 24 and Create Custom Gatherers
One of new features in JDK 24 release is Stream Gatherers (JEP 485). Stream Gatherers are a powerful enhancement to the Stream API that lets you build custom intermediate operations.The Gatherer interface in Java 24 allows for more complex intermediate operations in streams than were previously possible. It’s particularly useful for stateful operations that don’t fit well with map/filter/reduce patterns.
If you need to
- complex nested collectors
- multi-step stream operations
- custom new operations
- addition will enable you to manipulate data streams
Stream Gatherers provides new operations/features in addition to current Stream API.
- Gatherers.fold
Gatherers.fold is like a refined version of the Stream.reduce() method. Folding is a generalization of reduction. With reduction, the result type is the same as the element type, the combiner is associative, and the initial value is an identity for the combiner. For a fold, these conditions are not required, though we give up parallelizability.
For example with fold :
private static final List<String> stringStream = Arrays.asList("Paragraf 1","Paragraf 2","Paragraf 3","Paragraf 4","Paragraf 5","Paragraf 6");
var text2 = stringStream
.stream()
.gather(Gatherers.fold(() -> "", ((a, b) -> {
if(a.equals("")) return b.toString();
return a + ";" + b;
})))
.findFirst()
.get();
System.out.println("Gatherers text : " + text2);output is :
Gatherers text : Paragraf 1;Paragraf 2;Paragraf 3;Paragraf 4;Paragraf 5;Paragraf 6For example with reduce :
String result = stringStream.stream()
.reduce("", (a, b) -> a.isEmpty() ? b : a + ";" + b);
System.out.println("reduce : " + result);output is :
reduce : Paragraf 1;Paragraf 2;Paragraf 3;Paragraf 4;Paragraf 5;Paragraf 6You can use Gatherers with mixed type
For example :
public class MyObject implements Serializable {
private String strValue;
private Long longValue;
private Boolean boolValue;
private LocalDate dateValue;
public MyObject(String strValue, Long longValue, Boolean boolValue, LocalDate dateValue) {
this.strValue = strValue;
this.longValue = longValue;
this.boolValue = boolValue;
this.dateValue = dateValue;
}
}
//...
MyObject myObject1 = new MyObject("test",null,null,null);
MyObject myObject2 = new MyObject(null,100L,null,null);
MyObject myObject3 = new MyObject(null,null,Boolean.TRUE,null);
MyObject myObject4 = new MyObject(null,null,null, LocalDate.now());
var result = Stream.of(myObject1.getStrValue()
, myObject2.getLongValue()
, myObject3.getBoolValue()
, myObject4.getDateValue())
.gather(Gatherers.fold(() -> 0, (count, o) -> count + 1))
.findFirst()
.get();
System.out.println("mixed type : " + result);output is :
mixed type : 4Note : Performing a similar feat with reduce is difficult because the accumulator argument (count) is strongly typed
Stream.of(myObject1.getStrValue()
, myObject2.getLongValue()
, myObject3.getBoolValue()
, myObject4.getDateValue())
.reduce(0,(count,o) -> count +1) -> count +1 is bad operationCode get error : count +1 is bad operation
Using collectors is that more complex
var result2 = Stream.of(myObject1.getStrValue()
, myObject2.getLongValue()
, myObject3.getBoolValue()
, myObject4.getDateValue())
.collect(Collectors.toMap(obj -> obj, obj -> 1, Integer::sum, HashMap::new));
System.out.println("Collectors : " + result2.size());output is :
Collectors : 42. Gatherers.windowFixed
This function groups input elements into lists of a supplied size, emitting the windows downstream when they are full. Allows to separate elements in the stream into fixed groups.
For example :
private static final List<Integer> intStream = Arrays.asList(10,8,6,3,2,40,9);
output is :
[10, 8]
[6, 3]
[2, 40]
[9]For example :
private static final List<Integer> intStream = Arrays.asList(10,8,6,3,2,40,9,1);
Map<Integer,Integer> map = intStream.
stream().
gather(Gatherers.windowFixed(2)).collect(Collectors.toMap(integers -> integers.get(0), integers -> integers.get(1)));
map.forEach((integer, integer2) -> {
System.out.println("key : " + integer + " value : " + integer2);
});output is :
key : 2 value : 40
key : 6 value : 3
key : 9 value : 1
key : 10 value : 83. Gatherers.windowSliding
WindowSliding works like windowFixed() except each window starts on the next element in the source array, rather than at the end of the last window.Provides grouping of elements in a stream with floating elements of a specified size.
For example :
private static final List<Integer> intStream = Arrays.asList(10,8,6,3,2,40,9);
intStream.
stream().
gather(Gatherers.windowSliding(2))
.forEach(System.out::println);
System.out.println("----------------");
intStream.
stream().
gather(Gatherers.windowSliding(3))
.forEach(System.out::println);output is :
[10, 8]
[8, 6]
[6, 3]
[3, 2]
[2, 40]
[40, 9]
----------------
[10, 8, 6]
[8, 6, 3]
[6, 3, 2]
[3, 2, 40]
[2, 40, 9]4. Gatherers.scan
Scan is something like windowFixed but it accumulates the elements into a single element instead of an array.
For example
private static final List<Integer> intStream = Arrays.asList(10,8,6,3,2);
intStream.stream()
.gather(Gatherers.scan(() -> 100,
(current, next) -> current + next))
.forEach(System.out::println);
System.out.println("----------------");
int sum = intStream.stream()
.gather(Gatherers.scan(() -> 100,
(current, next) -> current + next))
.findFirst().get();
System.out.println("sum scan : " + sum);output is :
110
118
124
127
129
----------------
sum scan : 110Create Custom Gatherers Interface
You create a gatherer by defining four functions that work together that process input elements. Some of these functions are optional depending on your gatherer’s operation:
- initializer(): Creates the gatherer’s private state object
- integrator(): Integrates a new element from the input stream, possibly inspects the private state object, and possibly emits elements to the output stream
- combiner(): Combines two private state objects into one when the gatherer is processing the stream in parallel
- finisher(): Optionally performs an action after the gatherer has processed all input elements; it could inspect the private state object or emit additional output elements
Gatherer<T, A, R> gathererExample = Gatherer.of(
() -> ...., //initializer
(x,y) -> ....., //integrator
(s1,s2) -> ....., //combiner
(i,n) -> ..... , // finisher
);public record MyGatherer (int limit) implements Gatherer<Integer, List<Integer>, Integer> {
@Override
public Supplier<List<Integer>> initializer() {
return Gatherer.super.initializer();
}
@Override
public Integrator<List<Integer>, Integer, Integer> integrator() {
return null;
}
@Override
public BinaryOperator<List<Integer>> combiner() {
return Gatherer.super.combiner();
}
@Override
public BiConsumer<List<Integer>, Downstream<? super Integer>> finisher() {
return Gatherer.super.finisher();
}
}For example :
This example demonstrates Java 24’s new Gatherer interface in action. Let me explain the key components:
- BatchingGatherer (Class Implementation):
Takes a batch size parameter and collects elements until reaching that size
Contains required Gatherer methods: initializer(), integrator(), finisher(), and characteristics()
Produces lists of batched elements from the original stream
- BatchGatherer (Record Implementation):
Same functionality as BatchingGatherer but implemented as a record for conciseness
Shows how records can be used to create clean Gatherer implementations
- RunningAverageGatherer:
Another record-based Gatherer that calculates running averages
Demonstrates a different use case for Gatherers
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Gatherer;
import java.util.stream.Stream;
public class GathererExample {
public static void main(String[] args) {
List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
// Use our custom gatherer to create batches of numbers
List<List<Integer>> batches = numbers.stream()
.gather(new BatchingGatherer<>(3))
.toList();
System.out.println("Batches: " + batches);
// Use our custom gatherer with an alternate approach (using record)
List<List<Integer>> batches2 = numbers.stream()
.gather(BatchGatherer.ofSize(4))
.toList();
System.out.println("Batches (with record): " + batches2);
// Use running average gatherer
List<Double> runningAverages = numbers.stream()
.gather(RunningAverageGatherer.create())
.toList();
System.out.println("Running averages: " + runningAverages);
}
// class that implements the Gatherer interface to batch elements
static class BatchingGatherer<T> implements Gatherer<T, List<T>, List<T>> {
private final int batchSize;
public BatchingGatherer(int batchSize) {
this.batchSize = batchSize;
}
@Override
public Initializer<List<T>> initializer() {
return () -> new ArrayList<>();
}
@Override
public Integrator<List<T>, T, List<T>> integrator() {
return (batch, element, downstream) -> {
batch.add(element);
if (batch.size() >= batchSize) {
List<T> completedBatch = new ArrayList<>(batch);
batch.clear();
return downstream.push(completedBatch);
}
return true;
};
}
@Override
public Finisher<List<T>, List<T>> finisher() {
return (batch, downstream) -> {
if (!batch.isEmpty()) {
downstream.push(new ArrayList<>(batch));
}
return true;
};
}
@Override
public Characteristics characteristics() {
return Characteristics.of();
}
}
// record that implements the Gatherer interface
record BatchGatherer<T>(int batchSize) implements Gatherer<T, List<T>, List<T>> {
public static <T> BatchGatherer<T> ofSize(int size) {
return new BatchGatherer<>(size);
}
@Override
public Initializer<List<T>> initializer() {
return ArrayList::new;
}
@Override
public Integrator<List<T>, T, List<T>> integrator() {
return (batch, element, downstream) -> {
batch.add(element);
if (batch.size() >= batchSize) {
List<T> completedBatch = new ArrayList<>(batch);
batch.clear();
return downstream.push(completedBatch);
}
return true;
};
}
@Override
public Finisher<List<T>, List<T>> finisher() {
return (batch, downstream) -> {
if (!batch.isEmpty()) {
downstream.push(new ArrayList<>(batch));
}
return true;
};
}
}
// Another gatherer example for calculating running averages
record RunningAverageGatherer() implements Gatherer<Integer, RunningTotal, Double> {
public static RunningAverageGatherer create() {
return new RunningAverageGatherer();
}
@Override
public Initializer<RunningTotal> initializer() {
return () -> new RunningTotal(0, 0);
}
@Override
public Integrator<RunningTotal, Integer, Double> integrator() {
return (state, element, downstream) -> {
state.sum += element;
state.count++;
double average = (double) state.sum / state.count;
return downstream.push(average);
};
}
}
static class RunningTotal {
int sum;
int count;
RunningTotal(int sum, int count) {
this.sum = sum;
this.count = count;
}
}I hope this article is useful for you.
Thanx for reading
