{"id":2112,"date":"2026-02-07T01:41:06","date_gmt":"2026-02-06T12:41:06","guid":{"rendered":"https:\/\/www.ronella.xyz\/?p=2112"},"modified":"2026-02-07T01:41:07","modified_gmt":"2026-02-06T12:41:07","slug":"java-stream-gatherers","status":"publish","type":"post","link":"https:\/\/www.ronella.xyz\/?p=2112","title":{"rendered":"Java Stream Gatherers"},"content":{"rendered":"

Gatherers let you encode custom, often stateful, intermediate operations in a stream pipeline, going far beyond what map<\/code>, filter<\/code>, or flatMap<\/code> can express.<\/p>\n

\n

1. Why Gatherers Exist<\/h2>\n

A Gatherer<T, A, R><\/code> describes how elements of type T<\/code> flow through an intermediate stage, optionally using state A<\/code>, and emitting elements of type R<\/code>. <\/p>\n

    \n
  • It can perform one\u2011to\u2011one, one\u2011to\u2011many, many\u2011to\u2011one, or many\u2011to\u2011many transformations.<\/li>\n
  • It can maintain mutable state across elements, short\u2011circuit processing, and support parallel execution if given a combiner.<\/li>\n
  • You attach it using Stream.gather(gatherer)<\/code>.<\/li>\n<\/ul>\n

    This is analogous to Collector<\/code> for terminal operations, but acts mid\u2011pipeline<\/em> instead of at the end. <\/p>\n

    \n

    2. Gatherer.of<\/code> \u2013 Building Parallel\u2011Capable Gatherers<\/h2>\n

    You typically create gatherers using the static of<\/code> factory methods.<\/p>\n

    A key overload is: <\/p>\n

    static <T, A, R> Gatherer<T, A, R> of(\n        java.util.function.Supplier<A> initializer,\n        Gatherer.Integrator<A, T, R> integrator,\n        java.util.function.BinaryOperator<A> combiner,\n        java.util.function.BiConsumer<A, Gatherer.Downstream<? super R>> finisher\n)<\/code><\/pre>\n
    2.1 Arguments<\/h3>\n
      \n
    • Supplier<A> initializer<\/code> \u2013 creates the mutable state A<\/code> for each pipeline branch. <\/li>\n
    • Gatherer.Integrator<A, T, R> integrator<\/code> \u2013 per\u2011element logic; updates state, may push<\/code> outputs, and controls short\u2011circuit via its boolean return.<\/li>\n
    • BinaryOperator<A> combiner<\/code> \u2013 merges two states when running in parallel.<\/li>\n
    • BiConsumer<A, Downstream<? super R>> finisher<\/code> \u2013 flushes remaining state at the end of processing.<\/li>\n<\/ul>\n
      There are simpler overloads (e.g., stateless, no finisher) when you don\u2019t need all four.<\/p>\n
      2.2 Example: Custom map<\/code> Using Gatherer.of<\/code> (Stateless, Parallelizable)<\/h3>\n
      From the JDK docs, a gatherer equivalent to Stream.map<\/code> can be written as:<\/p>\n
      import java.util.function.Function;\nimport java.util.stream.Gatherer;\nimport java.util.stream.Stream;\n\npublic static <T, R> Gatherer<T, ?, R> map(Function<? super T, ? extends R> mapper) {\n    \/\/ stateless; state type is Void, no initializer\/combiner needed\n    return Gatherer.of(\n            (Gatherer.Integrator<Void, T, R>) (state, element, downstream) -> {\n                downstream.push(mapper.apply(element));\n                return true; \/\/ continue\n            }\n    );\n}\n\nvoid main() {\n    Stream.of("a", "bb", "ccc")\n          .gather(map(String::length))\n          .forEach(System.out::println);\n}<\/code><\/pre>\n
        \n
      • The gatherer is parallelizable<\/em> because we used of<\/code>, and it\u2019s stateless (Void<\/code> state).<\/li>\n
      • Rationale: for a simple one\u2011to\u2011one transformation, no state or finisher is needed; the integrator only pushes mapped elements.<\/li>\n<\/ul>\n
        \n
        3. Gatherer.ofSequential<\/code> \u2013 Sequential\u2011Only Gatherers<\/h2>\n
        For logic that is inherently sequential or where you don\u2019t care about parallel execution, you use ofSequential<\/code>.<\/p>\n
        Typical overloads:<\/p>\n
        static <T, R> Gatherer<T, Void, R> ofSequential(\n        Gatherer.Integrator<Void, T, R> integrator\n)\n\nstatic <T, A, R> Gatherer<T, A, R> ofSequential(\n        java.util.function.Supplier<A> initializer,\n        Gatherer.Integrator<A, T, R> integrator\n)\n\nstatic <T, A, R> Gatherer<T, A, R> ofSequential(\n        java.util.function.Supplier<A> initializer,\n        Gatherer.Integrator<A, T, R> integrator,\n        java.util.function.BiConsumer<A, Gatherer.Downstream<? super R>> finisher\n)<\/code><\/pre>\n
          \n
        • These gatherers are explicitly sequential; no combiner is provided and they are not used for parallel pipelines.<\/li>\n<\/ul>\n
          3.1 Example: Prefix Scan Using ofSequential<\/code><\/h3>\n
          The JDK docs show a prefix scan implemented with ofSequential<\/code>: <\/p>\n
          import java.util.function.BiFunction;\nimport java.util.function.Supplier;\nimport java.util.stream.Gatherer;\nimport java.util.stream.Stream;\n\npublic static <T, R> Gatherer<T, ?, R> scan(\n        Supplier<R> initial,\n        BiFunction<? super R, ? super T, ? extends R> scanner\n) {\n    class State {\n        R current = initial.get();\n    }\n\n    return Gatherer.<T, State, R>ofSequential(\n            State::new,\n            Gatherer.Integrator.ofGreedy((state, element, downstream) -> {\n                state.current = scanner.apply(state.current, element);\n                return downstream.push(state.current); \/\/ emit new prefix\n            })\n    );\n}\n\nvoid main() {\n    var numberStrings =\n            Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)\n                  .gather(scan(() -> "", (string, number) -> string + number))\n                  .toList();\n\n    System.out.println(numberStrings);\n}<\/code><\/pre>\n
            \n
          • Output: ["1", "12", "123", ... "123456789"]<\/code>.<\/li>\n
          • Rationale: prefix scan is inherently order\u2011sensitive and naturally modeled as sequential; ofSequential<\/code> expresses that contract directly.<\/li>\n<\/ul>\n
            \n
            4. Declaring a Sink Gatherer<\/h2>\n
            Example of a \u201clog\u2011only\u201d gatherer that never forwards elements:<\/p>\n
            import java.util.stream.Gatherer;\nimport java.util.stream.Stream;\n\npublic static Gatherer<String, ?, String> loggingSink() {\n    return Gatherer.ofSequential(\n            (Gatherer.Integrator<Void, String, String>) (state, element, downstream) -> {\n                System.out.println("LOG: " + element);\n                \/\/ Don't push anything downstream - just log and continue\n                return true;\n            }\n    );\n}\n\nvoid main() {\n    Stream.of("one", "two", "three")\n          .gather(loggingSink())\n          .forEach(s -> System.out.println("Downstream got: " + s)); \/\/ prints nothing downstream\n}<\/code><\/pre>\n
              \n
            • Here, the downstream will see nothing; the only observable effect is the logging side\u2011effect.<\/li>\n<\/ul>\n
              \n
              5. Built\u2011In Gatherer: windowSliding<\/code><\/h2>\n
              The Gatherers.windowSliding<\/code> method provides sliding windows as lists.<\/p>\n
              Signature: <\/p>\n
              static <T> java.util.stream.Gatherer<T, ?, java.util.List<T>> windowSliding(int windowSize)<\/code><\/pre>\n
              Behavior:<\/p>\n
                \n
              • Produces overlapping windows of size windowSize<\/code> in encounter order.  <\/li>\n
              • Each new window drops the oldest element and adds the next.  <\/li>\n
              • If the stream is empty, no windows; if shorter than windowSize<\/code>, one window containing all elements.  <\/li>\n<\/ul>\n
                5.1 Example: Sliding Windows of Integers<\/h3>\n
                import java.util.List;\nimport java.util.stream.Gatherers;\nimport java.util.stream.Stream;\n\nvoid main() {\n    List<List<Integer>> windows =\n            Stream.of(1, 2, 3, 4, 5, 6, 7, 8)\n                  .gather(Gatherers.windowSliding(3))\n                  .toList();\n\n    windows.forEach(System.out::println);\n}<\/code><\/pre>\n
                Expected result: [[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6], [5, 6, 7], [6, 7, 8]]<\/code>.<\/p>\n
                Rationale:<\/p>\n
                  \n
                • Sliding windows are a classic stateful pattern that require remembering the last windowSize - 1<\/code> elements.  <\/li>\n
                • Implementing this manually with map<\/code>\/flatMap<\/code> is error\u2011prone; windowSliding<\/code> encapsulates it as a reusable gatherer.<\/li>\n<\/ul>\n
                  \n
                  6. Built\u2011In Gatherer: mapConcurrent<\/code><\/h2>\n
                  mapConcurrent<\/code> applies a function concurrently using virtual threads while preserving stream order. Signature:<\/p>\n
                  static <T, R> java.util.stream.Gatherer<T, ?, R> mapConcurrent(\n        int maxConcurrency,\n        java.util.function.Function<? super T, ? extends R> mapper\n)<\/code><\/pre>\n
                  Behavior:<\/p>\n
                    \n
                  • Executes mapper<\/code> concurrently with up to maxConcurrency<\/code> in\u2011flight tasks.  <\/li>\n
                  • Uses virtual threads (Loom), so it scales well for blocking tasks.<\/li>\n
                  • Preserves encounter order when emitting results downstream.<\/li>\n
                  • Attempts to cancel in\u2011progress tasks when downstream no longer wants more elements.<\/li>\n<\/ul>\n
                    6.1 Example: Concurrent \u201cRemote\u201d Work<\/h3>\n
                    import java.util.List;\nimport java.util.stream.Gatherers;\nimport java.util.stream.Stream;\n\npublic class MapConcurrentDemo {\n\n    private static String fetchRemote(String id) {\n        try {\n            Thread.sleep(300); \/\/ simulate blocking IO\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n        }\n        return "response-for-" + id + " on " + Thread.currentThread();\n    }\n\n     static void main() {\n        List<String> responses =\n                Stream.of("A", "B", "C", "D", "E")\n                      .gather(Gatherers.mapConcurrent(3, MapConcurrentDemo::fetchRemote))\n                      .toList();\n\n        responses.forEach(System.out::println);\n    }\n}<\/code><\/pre>\n
                      \n
                    • Up to 3 virtual threads will run fetchRemote<\/code> concurrently.<\/li>\n
                    • The result list preserves the order "A", "B", "C", "D", "E"<\/code>.<\/li>\n<\/ul>\n
                      Rationale:<\/p>\n
                        \n
                      • Compared to parallel()<\/code>, mapConcurrent<\/code> is explicit about concurrency level, suited for blocking IO, and guarantees order, making it a better fit for many modern workloads.<\/li>\n<\/ul>\n
                        \n
                        7. Putting It All Together<\/h2>\n
                        You now have:<\/p>\n
                          \n
                        • Gatherer.of<\/code><\/strong> to build parallel\u2011capable gatherers when you need full control, including a combiner and finisher.<\/li>\n
                        • Gatherer.ofSequential<\/code><\/strong> for simpler or inherently sequential logic, with examples like prefix scan. <\/li>\n
                        • Gatherers.windowSliding<\/code><\/strong> and Gatherers.mapConcurrent<\/code><\/strong> as high\u2011level, ready\u2011made gatherers for windowing and concurrent mapping.<\/li>\n<\/ul>\n
                          With these building blocks, you can design expressive, stateful, and performance\u2011aware stream pipelines using the latest Java Stream API.<\/p>\n","protected":false},"excerpt":{"rendered":"
                          Gatherers let you encode custom, often stateful, intermediate operations in a stream pipeline, going far beyond what map, filter, or flatMap can express. 1. Why Gatherers Exist A Gatherer<T, A, R> describes how elements of type T flow through an intermediate stage, optionally using state A, and emitting elements of type R. It can perform […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[93],"tags":[],"_links":{"self":[{"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=\/wp\/v2\/posts\/2112"}],"collection":[{"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=2112"}],"version-history":[{"count":1,"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=\/wp\/v2\/posts\/2112\/revisions"}],"predecessor-version":[{"id":2113,"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=\/wp\/v2\/posts\/2112\/revisions\/2113"}],"wp:attachment":[{"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=2112"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=2112"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.ronella.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=2112"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}