Stream Gatherers in Practice
Write your own gatherers for back-end jobs: collapsing duplicates, grouping runs, stopping at a budget, and calling services with bounded concurrency.
Gatherer.<T, State, R>ofSequential(
State::new, // initializer
Gatherer.Integrator.ofGreedy((s, t, down) -> ...),
(s, down) -> { /* flush */ })Stream.gather has been final since Java 24 (JEP 485). The built-in Gatherers cover windows, running totals and concurrent mapping. In real services you soon need an operation that is specific to your data, and a small custom gatherer is often clearer than a loop with flags.
Anatomy of a gatherer
| Part | Role |
|---|---|
| initializer | Creates fresh mutable state for each stream evaluation |
| integrator | Receives state, element and Downstream; pushes zero or more results; returns false to stop |
| combiner | Merges states for parallel streams (only with Gatherer.of) |
| finisher | Runs once at the end; flushes anything still held in state |
Gatherer.ofSequential(initializer, integrator, finisher) is the factory to start with. Wrap the integrator in Gatherer.Integrator.ofGreedy when it never stops early, which lets the stream skip some short-circuit checks.
Rules that keep gatherers correct
- State lives in the initializer. A field or captured list shared across evaluations breaks as soon as the same gatherer is used twice, or by two requests at once.
- Respect the downstream.
downstream.push(x)returnsfalsewhen no more elements are wanted, for example afterlimit. Return that value from the integrator. In the finisher, checkdownstream.isRejecting()before pushing. - Flush in the finisher. A gatherer that buffers (a batch, a group of consecutive items) loses the last group if it has no finisher.
- Short-circuit on purpose. Returning
falsefrom the integrator ends the stream, which makes infinite sources such as paginated APIs safe to consume.
Composing
a.andThen(b) joins two gatherers into one, so a pipeline step such as "collapse duplicates, then batch" can be named, tested once and reused across jobs.
mapConcurrent for I/O
Gatherers.mapConcurrent(max, fn) runs fn on virtual threads with at most max in flight and emits results in input order. It suits calling a downstream service for each ID. Pick max from what that service tolerates, not from your CPU count. The function runs on other threads, so a ScopedValue binding or ThreadLocal from the caller is not visible inside it; pass what it needs as data.
Mistakes people make
- Writing a gatherer when
filter,mapor a collector already does the job. - Pushing results from the finisher without checking
isRejecting(). - Holding unbounded state, such as every element seen so far, on a stream fed from a large table.
Example
When to use it
- Collapsing duplicate webhook deliveries before writing changes to the database in batches.
- Reading a paginated upstream API lazily and stopping once a size or time budget is used up.
- Enriching a list of order IDs with prices from a pricing service while capping concurrent requests.
More examples
Grouping consecutive log lines by request
The integrator emits a group whenever the key changes, and the finisher flushes the final run that is still in state. Unlike <code>groupingBy</code>, runs with the same key stay separate, which is what you want for interleaved logs.
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Gatherer;
record Line(String requestId, String text) {}
record Group(String key, int lines) {}
static <T> Gatherer<T, ?, Group> runsBy(Function<T, String> key) {
class Run { String key; int count; }
return Gatherer.ofSequential(
Run::new,
(run, item, downstream) -> {
String k = key.apply(item);
if (run.count > 0 && !k.equals(run.key)) {
if (!downstream.push(new Group(run.key, run.count))) return false;
run.count = 0;
}
run.key = k;
run.count++;
return true;
},
(run, downstream) -> {
if (run.count > 0 && !downstream.isRejecting()) {
downstream.push(new Group(run.key, run.count));
}
});
}
void main() {
List<Line> log = List.of(
new Line("a1", "GET /orders"), new Line("a1", "db 4ms"), new Line("a1", "200"),
new Line("b7", "POST /login"), new Line("b7", "401"),
new Line("a1", "late callback"));
log.stream().gather(runsBy(Line::requestId)).forEach(g -> IO.println(g));
}
// Output:
// Group[key=a1, lines=3]
// Group[key=b7, lines=2]
// Group[key=a1, lines=1]Stopping at a byte budget on an endless source
Returning <code>false</code> from the integrator ends an infinite stream once the next page would exceed the budget. Page 4 (700 bytes) would take the total to 2,200, so it is dropped and no later pages are requested.
import java.util.stream.Gatherer;
import java.util.stream.Stream;
record Page(int number, int bytes) {}
static Gatherer<Page, ?, Page> withinBudget(long maxBytes) {
class Used { long bytes; }
return Gatherer.ofSequential(
Used::new,
(used, page, downstream) -> {
if (used.bytes + page.bytes() > maxBytes) return false; // stop the stream
used.bytes += page.bytes();
return downstream.push(page);
});
}
void main() {
// Simulates an API that always has another page.
Stream<Page> pages = Stream.iterate(1, n -> n + 1).map(n -> new Page(n, 300 + n * 100));
pages.gather(withinBudget(2_000))
.forEach(p -> IO.println("page " + p.number() + " (" + p.bytes() + " bytes)"));
}
// Output:
// page 1 (400 bytes)
// page 2 (500 bytes)
// page 3 (600 bytes)Bounded concurrent calls with mapConcurrent
Six lookups run with at most three requests in flight, and the results come back in input order. Catching <code>IOException</code> inside the function turns one failed call into a value instead of failing the whole stream, while an interrupt is restored rather than swallowed.
import module jdk.httpserver;
import module java.net.http;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Gatherers;
void main() throws Exception {
AtomicInteger inFlight = new AtomicInteger();
AtomicInteger peak = new AtomicInteger();
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
server.createContext("/price/", exchange -> {
peak.accumulateAndGet(inFlight.incrementAndGet(), Math::max);
try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
String id = exchange.getRequestURI().getPath().substring(7);
byte[] body = (id + "=" + id.length() * 10).getBytes(StandardCharsets.UTF_8);
inFlight.decrementAndGet();
exchange.sendResponseHeaders(200, body.length);
try (OutputStream out = exchange.getResponseBody()) { out.write(body); }
});
server.start();
String base = "http://127.0.0.1:" + server.getAddress().getPort() + "/price/";
try (HttpClient client = HttpClient.newHttpClient()) {
List<String> prices = Stream.of("a", "bb", "ccc", "dddd", "eeeee", "ffffff")
.gather(Gatherers.mapConcurrent(3, id -> {
try {
var req = HttpRequest.newBuilder(URI.create(base + id)).build();
return client.send(req, HttpResponse.BodyHandlers.ofString()).body();
} catch (IOException e) {
return id + "=error";
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return id + "=cancelled";
}
}))
.toList();
IO.println(prices);
IO.println("never more than 3 at once: " + (peak.get() <= 3));
} finally {
server.stop(0);
}
}
// Output:
// [a=10, bb=20, ccc=30, dddd=40, eeeee=50, ffffff=60]
// never more than 3 at once: true
Discussion