企业官网建设流程全解析

• 后端
• 并发编程
• 异步编程

导读

StreamConverters.javaCollectorParallelUnordered是 Akka Stream 提供的一个将 Java 8Collector以并行方式接入响应式流的 Sink 操作符。它基于图阶段Balance将上游元素分发到多个异步 worker 中分别累积，再用Collector.combiner归约合并，最终物化为Future（Scala）或CompletionStage（Java）。读完本文，你将掌握该操作符的精确 API 签名、并行执行的数据流拓扑、与顺序版javaCollector的取舍，以及可复制的使用示例和仓库内的验证用例。

操作符概览

根据 操作符参考文档，该操作符的功能定位是：

创建一个 Sink，它会物化为一个 @scala[Future] @java[CompletionStage]，并在其中完成 Java 8Collector的转换（transformation）与归约（reduction）操作的结果。

换句话说，它把标准 Java 8java.util.stream.Collector的能力带入了 Akka Stream 的响应式管线：上游的元素流入该 Sink 后被累积进可变的中间结果容器，所有元素处理完毕后，再经由Collector的可选 finisher 变换成最终结果，整个过程的产物是一个异步完成的结果句柄。它属于StreamConverters工具对象下 Additional Sink and Source converters 这一组转换器家族，与 javaCollector、asJavaStream、fromJavaStream、asInputStream等并列。

与顺序版的javaCollector不同，这里的归约处理是基于图Balance并行执行的（"Reduction processing is performed in parallel based on graphBalance"，见 scaladsl/StreamConverters.scala 中的源码注释），因此适合元素量大、累积计算相对重的场景。

API 签名

原文档给出的签名如下（出自 javaCollectorParallelUnordered.md）：

Scala（scaladsl.StreamConverters）：

def javaCollectorParallelUnorderedT, R( collectorFactory: () => java.util.stream.Collector[T, _ <: Any, R]): Sink[T, Future[R]]

Java（javadsl.StreamConverters）：

def javaCollectorParallelUnorderedT, R( collector: akka.japi.function.Creator[Collector[T, _ <: Any, R]]): Sink[T, CompletionStage[R]]

关键参数语义：

| 参数 | 类型 | 含义 | | -- | -- | -- | |parallelism|Int| 并行分片的数量，即图中Balance的输出端口数，也就是同时进行累积的 worker 个数 | |collectorFactory/collector|() => Collector[T, _, R]/Creator[Collector[T, _, R]]| 一个工厂函数，每次需要时生产一个新的Collector实例；传入的必须是工厂而非Collector本身（详见下文"注意事项"） |

类型参数T是流入元素的类型，R是最终归约结果的类型；中间累积容器类型A（即Collector[T, A, R]中的A）被泛型擦除为Any，由内部状态类持有。Java 侧传入akka.japi.function.Creator后，内部会包装成 Scala 的函数字面量() => collector.create()，最终得到CompletionStage[R]形式的物化值（通过.toCompletionStage()转换，见 javadsl/StreamConverters.scala）。

并行执行的实现原理

从源码结构看，parallelism > 1时该 Sink 的图拓扑是一个经典的"分而治之"结构（实现位于 scaladsl/StreamConverters.scala）：

1. 上游元素进入BalanceT：一个公平分发的扇出阶段，把元素轮流/按需分配到parallelism个输出端口；
2. 每个端口连接一个worker 流水线：Flow[T].fold(...).async，即以fold形式做局部的顺序累积（把每个元素state.update(elem)进CollectorState），并用.async标记异步边界，使各 worker 可并行推进；
3. 各 worker 的输出汇入Merge[CollectorState[T, R]](parallelism)；
4. Merge下游再叠加一个fold，使用ReducerState对各个分片的累积结果做归约——这一步调用的是Collector的combiner函数（BinaryOperator<A>），把"多批局部结果"两两合并；
5. 归约完成后执行state.finish()（调用Collector.finisher）得到最终结果R，交给Sink.head[R]，由此物化出一个在流完成时完成的Future[R]。

这个 Sink 的默认属性名为javaCollectorParallelUnordered，注册在 impl/Stages.scala 中。

内部的 CollectorState 与 ReducerState

并行收集的关键在于两套内部状态类（定义于 impl/Sinks.scala，均为@InternalApi private[akka]）：

• CollectorState[T, R]：负责"累积"一侧。FirstCollectorState在收到第一个元素时才调用collectorFactory()创建真正可变的Collector，取出supplier().get()得到累积容器、accumulator()得到累积函数，然后用accumulator.accept(accumulated, elem)累积元素；之后的元素交给MutableCollectorState原地更新，finish()时用finisher().apply(...)收尾。把工厂调用延迟到首个元素到达、且每次update都返回新实例，是为了保证不同 materialization 之间绝不共享同一个可变 Collector。
• ReducerState[T, R]：负责"归约"一侧。FirstReducerState收到第一批局部结果时取出collector.combiner()，之后MutableReducerState.update反复执行reduced = combiner(reduced, batch)；空流情况下finish()会以null作为累积值调用 finisher（collector.finisher().apply(null)）。

由此可见，该操作符能否正确工作，强依赖于你提供的Collector本身是可组合的：它必须实现了有意义的supplier、accumulator、combiner与finisher，其中combiner是并行版本独有的、串行版本根本不会触碰的组件。

parallelism == 1 时的退化行为

源码中有一个容易忽略的重要细节（scaladsl/StreamConverters.scala）：

if (parallelism == 1) javaCollectorT, R else { ... }

当parallelism == 1时，javaCollectorParallelUnordered会直接委托给顺序版javaCollector，走Flow.fold+Sink.head的简单管线，不再构建Balance/Merge图。这保证了：即使调用方传 1，也不会出现"并行度 1 却绕一圈分片归约"的额外开销。也正因如此，parallelism的合法下限是 1，传入 0 或负数将没有意义（会进入 else 分支构造出端口数非法/无意义的图），实践中应从 2 开始体现并行收益。

与顺序版 javaCollector 的对比与取舍

| 维度 |javaCollector|javaCollectorParallelUnordered| | -- | -- | -- | | 归约方式 | 单条流水线顺序累积（"Reduction processing is performed sequentially"） | 基于Balance并行累积，combiner归约（"performed in parallel based on graphBalance"） | | 物化值 |Future[R]/CompletionStage[R]| 相同 | | 是否使用combiner| 不使用 | 使用 | | 并行度参数 | 无 |parallelism: Int（为 1 时退化为顺序版） | | 元素到达最终结果的次序 | 累积有序 |无序（"Unordered"——多 worker 各自的局部结果以任意顺序到达Merge并被 combiner 合并） |

两条实现的事实依据分别见 scaladsl/StreamConverters.scala（顺序版）与同文件 L121-L162（并行版）。选择建议：元素量大、且你的Collector具备廉价且正确的combiner（如Collectors.summingInt、Collectors.toList、Collectors.joining等 JDK 标准实现）时，用并行版摊薄累积成本；对顺序敏感或依赖流的固有次序做折叠时，用顺序版。

实战示例

仓库的文档代码示例位于 JavaCollectorDocExample.scala（顺序版，供对照）与 JavaCollectorDocExamples.java。并行版可以直接按如下方式使用：

Scala：

import java.util.stream.Collectors import akka.stream.scaladsl.{ Source, StreamConverters } import akka.actor.ActorSystem implicit val system: ActorSystem = ActorSystem("demo") // 并行收集为 List（注意 parallelism=4） val future: scala.concurrent.Future[java.util.List[String]] = Source(List("one", "two", "three")) .runWith(StreamConverters.javaCollectorParallelUnordered(4)(() => Collectors.toList[String]()))

Java：

import akka.stream.javadsl.Source; import akka.stream.javadsl.StreamConverters; import java.util.concurrent.CompletionStage; import java.util.stream.Collectors; Source.from(java.util.List.of("one", "two", "three")) .runWith(StreamConverters.javaCollectorParallelUnordered(4, Collectors::toList), system);

并行求和（仓库测试中的写法，见 StreamConvertersSpec.scala）：

val future = Source(1 to 100) .runWith(StreamConverters.javaCollectorParallelUnordered(4)(() => Collectors.summingIntInt)) future.futureValue.toInt should ===(5050) // 1+2+...+100

注意 Java 8 的Collector接口本身对collector的Characteristics（如CONCURRENT、UNORDERED、IDENTITY_FINISH）有一定语义约定；Akka 的实现默认按"可并发累积、无序归约"的方式分片使用，因此像Collectors.joining(", ")这类实现会通过combiner得到拼接结果，但拼接次序不保证与上游元素原始顺序一致——这是名字里 "Unordered" 的含义，若业务对顺序敏感需自行权衡。

使用注意事项

1. 必须传工厂而非实例：collectorFactory/Creator会被多次调用——每次流 materialization 都会重建状态，且在并行图里每个 worker 各自需要一份 Collector。源码注释明确提醒："a flow can be materialized multiple times, so the function producing theCollectormust be able to handle multiple invocations"（scaladsl/StreamConverters.scala）。若直接复用同一个可变Collector，会造成状态跨流共享、结果错误。
2. 工厂调用是惰性的：FirstCollectorState在收到首个元素时才实例化 Collector；对空流，finish()会单独调用一次工厂并直接对空容器应用 finisher（见 impl/Sinks.scala）。测试用例 "work parallelly with an empty source" 验证了空流下joining(", ")得到""（StreamConvertersSpec.scala）。
3. 可复用性已验证：同一 Sink 可被多次runWith，且各次结果互不影响——仓库测试 "be reusable with parallel version" 用同一个javaCollectorParallelUnordered(4)(...)先对 1..4 求和得 10、再对 4..6 求和得 15，印证工厂模式隔离了状态（StreamConvertersSpec.scala）。
4. 异常传播：supplier/accumulator/combiner/finisher中抛出的异常会沿流水线传播并导致物化出的Future/CompletionStage以失败结束（对应测试见 StreamConvertersSpec.scala 附近）。

验证与测试

仓库内对javaCollectorParallelUnordered的覆盖测试位于 akka-stream-tests/src/test/scala/akka/stream/scaladsl/StreamConvertersSpec.scala，核心断言包括：

• Source(1 to 100)+Collectors.summingInt，parallelism = 4，最终结果为 5050（验证并行累积+combiner 归约的正确性）；
• 空源 +Collectors.joining结果为""（验证空流路径与 finisher 对空容器的处理）；
• Sink 复用场景下两次runWith分别得到 10 与 15（验证工厂隔离、无跨 materialization 状态泄漏）。

这些用例同时是理解该操作符行为边界的可直接运行的参考。

相关参考

• 操作符文档：javaCollectorParallelUnordered.md
• 顺序版对照文档：javaCollector.md
• 转换器家族索引：operators/index.md
• Scala 实现：scaladsl/StreamConverters.scala
• Java 实现：javadsl/StreamConverters.scala
• 内部状态实现：impl/Sinks.scala
• 默认属性注册：impl/Stages.scala
• 行为验证测试：StreamConvertersSpec.scala

• 后端
• 并发编程
• 异步编程