企业官网建设流程全解析

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

导读

Source.single是 Akka Streams 中用于创建一个只发射单个元素、随后立即完成的 Source 的最基础工厂方法。无论是 Scala 还是 Java API，它都以极低的成本为后续流式处理链提供一个"种子"值，常用于测试、模拟单条消息、触发一次异步查询等场景。读完本文，你将掌握Source.single的精确签名、Reactive Streams 背压语义、Scala/Java 双语言示例，并能从源码层面理解其内部SingleSourceGraphStage 的实现原理及其在FlattenMerge等场景中的专门优化。

概览与定位

Source.single属于 Akka Streams 的 Source 操作符家族，其核心行为可以概括为：

• 只发射一次：把给定的单个对象作为唯一元素向下游推送；
• 发射后立即完成：元素被下游接收后，流随即进入 completed 状态；
• 无外部副作用：素材化（materialize）得到的类型为NotUsed，不携带可管理的资源句柄。

它是理解 Akka Streams 中"有限流"与"一次性发射"语义的最佳入门操作符，也是构建更复杂数据流的最小积木。

Signature（签名）

Source.single在 Scala 与 Java 两个 API 面上的签名如下：

• Scala：singleT: akka.stream.scaladsl.Source[T, akka.NotUsed]
• Java：single(T)

素材化值的类型是NotUsed，说明这个 Source 本身不产生有意义的素材化结果；如果你需要拿到发射的元素，应通过下游的Sink.head、Sink.seq等操作符来收集。

行为描述

Source.single将给定的单个对象流式发射一次，发射完成后流立即结束。每一个连接到该 Source 的 Sink 都会看到各自独立的一条"只含一个元素"的流——这源于 Akka Streams 的图（Graph）可以被多次素材化的特性：每次run都是一次全新的物化，都会重新走一遍SingleSource的onPull → push → complete流程，因此不同次运行之间互不影响，同一个 Source 可以安全地反复使用。

相关操作符对比

Source.single经常与以下三个操作符放在一起比较，它们在"重复/周期性发射"上的行为截然不同：

典型的选择依据：只需要一个值用single；需要无限重复同一值用repeat（配合take(n)限定数量）；需要按时间周期触发用tick；需要循环遍历一组元素用cycle。

示例

以下示例均取自仓库内的真实测试代码，可直接运行验证。

Scala 示例

来自 SourceSpec.scala：

import akka.stream._ import akka.NotUsed val s: Future[immutable.Seq[Int]] = Source.single(1).runWith(Sink.seq) s.foreach(list => println(s"Collected elements: $list")) // prints: Collected elements: List(1)

Java 示例

来自 SourceTest.java：

import akka.stream.javadsl.Source; import akka.stream.javadsl.Sink; CompletionStage<List<String>> future = Source.single("A").runWith(Sink.seq(), system); CompletableFuture<List<String>> completableFuture = future.toCompletableFuture(); completableFuture.thenAccept(result -> System.out.printf("collected elements: %s\n", result)); // result list will contain exactly one element "A"

对应的测试断言（SourceSpec.scala 与 SourceTest.java）分别验证了 Scala 侧收集到immutable.Seq(1)、Java 侧结果列表大小为 1 且元素为"A"，从测试层面确认了"恰好一个元素"的行为契约。

与 Sink.head 组合

如果只想取这一个值（而不是收集成 Seq/List），可以搭配Sink.head：

val one: Future[Int] = Source.single(42).runWith(Sink.head)

由于single只发射一个元素，Sink.head一定能拿到值并正常完成，不会出现NoSuchElementException。

Reactive Streams 语义

Source.single的 Reactive Streams 语义非常简洁，可用下表概括：

正因为"只发射一次 + 完成后立即停止"，它天然满足 Reactive Streams 规范中"订阅后最多发射 N 个元素、随后正常完成"的有界语义，不需要任何额外的资源清理。

源码剖析：SingleSource 的实现原理

工厂方法入口

在 Source.scala 中，single的实现非常轻量：

/** * Create a `Source` with one element. * Every connected `Sink` of this stream will see an individual stream consisting of one element. */ def singleT: Source[T, NotUsed] = fromGraph(new GraphStages.SingleSource(element))

它直接包装了一个内部 GraphStage——GraphStages.SingleSource，没有额外的分配与转换开销。

GraphStage 核心逻辑

SingleSource定义在 GraphStages.scala：

final class SingleSourceT extends GraphStage[SourceShape[T]] { override def initialAttributes: Attributes = DefaultAttributes.singleSource ReactiveStreamsCompliance.requireNonNullElement(elem) val out = OutletT val shape = SourceShape(out) def createLogic(attr: Attributes) = new GraphStageLogic(shape) with OutHandler { def onPull(): Unit = { push(out, elem) completeStage() } setHandler(out, this) } override def toString: String = "SingleSource" }

几个关键实现细节值得注意：

1. 构造时非空校验：ReactiveStreamsCompliance.requireNonNullElement(elem)在构建阶段就拒绝null元素，从源头保证了 Reactive Streams 规范中"禁止发射 null 元素"的要求；
2. 拉取驱动（pull-based）：下游发出需求（onPull）后，SingleSource才执行push(out, elem)发射元素，紧接着调用completeStage()完成整个 stage——这也正是文档中"发射一次、完成后结束"语义的直接代码体现；
3. 属性标记：它带有DefaultAttributes.singleSource这一初始属性，便于上层对这类特殊 Source 做识别与优化。

专门的性能优化：FlattenMerge 中的单元素捷径

SingleSource不仅在语义上特殊，在实现层面还有专门的优化路径。在 TraversalBuilder.scala 中提供了getSingleSource工具方法，用于在图遍历构建阶段直接识别出SingleSource（或仅包裹了它、且未做异步/素材化改造的线性图），从而在FlattenMerge（扁平化合并内部流）等场景中跳过子流的完整物化过程，直接把元素推入下游队列，避免为单个元素创建完整的子流运行环境，显著降低开销。

相关逻辑同样体现在 StreamOfStreams.scala：队列中可以持有SubSinkInlet[T]或SingleSource，当识别到SingleSource时直接推送元素而非物化子流。从源码结构看，这是 Akka Streams 为"单元素 Source"这一高频基础场景所做的专门性能设计，也解释了为何Source.single会成为流式编程中极低成本的"种子源"。

典型应用场景

结合上述语义与实现，Source.single的典型应用场景包括：

• 测试与模拟：仓库的SourceSpec/SourceTest中大量用它构造确定性的单元素流来验证下游行为；
• 作为流式管道起点：把一个外部计算得到的值灌入流式处理链（map、filter、flatMapConcat等）继续加工；
• 触发一次异步副作用：结合mapAsync对单个请求执行一次异步调用，Sink.seq收集唯一结果；
• 作为FlattenMerge的内层元素：受益于上文所述的单元素捷径优化，以极低开销参与流的扁平化合并。

小结

Source.single用最简洁的接口封装了"单元素、单次发射、发射即完成"这一基础流式语义：工厂方法在 Source.scala 中一行实现，底层由 GraphStages.scala 中的SingleSource承担"拉取即推送、推送即完成"的执行逻辑，并由 TraversalBuilder.scala 提供针对性的物化优化。无论是学习 Akka Streams 的背压与完成语义，还是在真实项目中构造单值数据流，Source.single都是应当首先掌握的基础操作符。

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