Claude
6b5a918278
gatherers/: windowFixed, windowSliding, fold, scan, mapConcurrent, a custom Gatherer (dedupeConsecutive, takeUntil), and a Collector-vs-Gatherer comparison. Verified: windowSliding emits one short window on a too-short stream rather than shrinking to empty like windowFixed; mapConcurrent waits for in-flight siblings to finish (~901ms) rather than cancelling them when one mapper throws. scoped-values/: ScopedValue API and rebinding, why plain threads (virtual or platform) do not inherit a binding while a StructuredTaskScope subtask does, and the memory cost vs InheritableThreadLocal measured two ways -- a noisy heap-delta first pass, then a precise jcmd GC.class_histogram object census (5 shared Carrier objects vs ~700,000 copied ThreadLocalMap/Entry objects for 100,000 threads). Companion code for the ankurm.com posts "Java Stream Gatherers (JEP 485)" and "Scoped Values vs ThreadLocal in Java 25: Migration Guide with Virtual Threads". Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01FtpJvZfg4nvLvtzgJTDWpB
67 lines
4.9 KiB
Markdown
67 lines
4.9 KiB
Markdown
# 16. Stream Gatherers (JEP 485, final since JDK 24)
|
|
|
|
Prev: [15. Build files](15-build-files.md) · Next: [17. Scoped values vs ThreadLocal](17-scoped-values-migration.md)
|
|
|
|
Compiles and runs on JDK 25+ with **no** `--enable-preview` flag -- JEP 485 finalized in JDK 24 and shipped unchanged in 25.
|
|
Source: [`gatherers/src/`](../gatherers/src), transcripts [98-basics](output/98-basics.txt), [98-mapconcurrent](output/98-mapconcurrent.txt),
|
|
[98-custom-gatherer](output/98-custom-gatherer.txt), [98-collector-vs-gatherer](output/98-collector-vs-gatherer.txt).
|
|
|
|
## The five built-in gatherers ([`GathererBasics.java`](../gatherers/src/GathererBasics.java), [`MapConcurrentDemo.java`](../gatherers/src/MapConcurrentDemo.java))
|
|
|
|
```java
|
|
Stream.of(1,2,3,4,5,6,7).gather(Gatherers.windowFixed(3)).toList(); // [[1,2,3],[4,5,6],[7]]
|
|
Stream.of(1,2,3,4,5).gather(Gatherers.windowSliding(3)).toList(); // [[1,2,3],[2,3,4],[3,4,5]]
|
|
Stream.of(1,2,3,4).gather(Gatherers.fold(() -> "", (a,n) -> a+"["+n+"]")).findFirst(); // "[1][2][3][4]"
|
|
Stream.of(1,2,3,4,5).gather(Gatherers.scan(() -> 0, Integer::sum)).toList(); // [1,3,6,10,15]
|
|
Stream.of(1,2,3,4).gather(Gatherers.mapConcurrent(4, Worker::call)).toList(); // order preserved
|
|
```
|
|
|
|
* `windowFixed(n)` throws `IllegalArgumentException` for `n <= 0` ("'windowSize' must be greater than zero"), not a silent empty result.
|
|
* `windowSliding(n)` does **not** drop a too-short remainder the way `windowFixed` shrinks its last window -- on a
|
|
stream shorter than `n`, it still emits one window holding whatever is there (`[1,2]` for `windowSliding(3)` over
|
|
`[1,2]`), and only an empty *source* stream produces zero windows. Easy to get backwards if you assume it mirrors
|
|
`windowFixed`'s "shrink the leftover" rule.
|
|
* `fold` and `scan` share a signature (`Supplier<R>`, `BiFunction<R,T,R>`) but not a shape: `fold` emits exactly one
|
|
final value (a stream of size 1), `scan` emits every intermediate accumulation. Neither has a combiner, so neither
|
|
runs in parallel -- both are sequential-only by design, which is what lets `R` differ from `T` in the first place
|
|
(`Stream.reduce` can't do that safely in parallel without a combiner).
|
|
* `mapConcurrent(max, fn)` runs each element's mapper on its **own** virtual thread (confirmed via
|
|
`Thread.currentThread().isVirtual()`), bounded to `max` in flight at once, output in encounter order regardless of
|
|
completion order. **The one behavior worth isolating**: a mapper that throws does **not** cancel its still-running
|
|
siblings. [98-mapconcurrent.txt](output/98-mapconcurrent.txt) has element 3 throw immediately while elements
|
|
1/2/4/5 each sleep 900ms -- the exception doesn't surface until ~900ms later, once every already-started mapper has
|
|
finished. `mapConcurrent` is not `StructuredTaskScope`; there's no fail-fast cancellation here.
|
|
|
|
## Writing a custom Gatherer ([`CustomGatherer.java`](../gatherers/src/CustomGatherer.java))
|
|
|
|
`Gatherer<T,A,R>` is four composable pieces -- initializer (private state `A`), integrator (per element: touch state,
|
|
push 0+ `R` downstream, return whether to keep pulling), combiner (parallel merge, omit for sequential-only), finisher
|
|
(flush on exhaustion). `Gatherer.ofSequential(...)` skips the combiner entirely.
|
|
|
|
```java
|
|
static <T> Gatherer<T, ?, T> dedupeConsecutive() {
|
|
return Gatherer.ofSequential(() -> new ArrayList<Object>(List.of()), (state, element, downstream) -> {
|
|
boolean changed = state.isEmpty() || !state.get(0).equals(element);
|
|
if (changed) { if (state.isEmpty()) state.add(element); else state.set(0, element); return downstream.push(element); }
|
|
return true;
|
|
});
|
|
}
|
|
```
|
|
|
|
`takeUntil` in the same file is the short-circuiting case a `Collector` structurally cannot express: the integrator
|
|
returns `false` once the stop condition is met, and upstream is never pulled again --
|
|
[98-custom-gatherer.txt](output/98-custom-gatherer.txt) proves it with a `peek()` upstream of the gather that only
|
|
ever sees the 3 elements actually consumed, not all 6 in the source.
|
|
|
|
## Collector vs Gatherer ([`CollectorVsGatherer.java`](../gatherers/src/CollectorVsGatherer.java))
|
|
|
|
A `Collector` always reduces to exactly one terminal value and can only sit at the end of a pipeline. A `Gatherer` is
|
|
an intermediate op -- it returns `Stream<R>`, so a gather can be followed by `filter`, `map`, another `gather`, or a
|
|
`collect`. `groupingBy`/`summingDouble` for an order-totals `Map` is still the right call in
|
|
[98-collector-vs-gatherer.txt](output/98-collector-vs-gatherer.txt): the answer really is one `Map`, and reaching for
|
|
`gather()` there would add ceremony with nothing to show for it. `scan` feeding straight into `filter`/`map` in the
|
|
same lazy pipeline is the case a `Collector` can't do without materializing an intermediate `List` and starting a
|
|
second stream.
|
|
|
|
Prev: [15. Build files](15-build-files.md) · Next: [17. Scoped values vs ThreadLocal](17-scoped-values-migration.md)
|