scala - Akka 流 : Why does Sink. head 终止流 alsoTo 广播？

Question

让我们来看一个非常简单的案例:

Source(1 to 10)
  .alsoTo(Sink.foreach(v => println(s"each: $v")))
  .toMat(Sink.head)(Keep.right)
  .run()

根据 alsoTo 文档，我希望 Sink.foreach 打印所有元素，但是，它只打印第一个。如果我切换 Sink.foreach 和 Sink.head 位置，也会发生同样的情况。

但是，如果广播是通过 GraphDSL 实现的，即使其中一个接收器是 Sink.head，也会消耗整个源。

编辑:alsoTo 的文档说明如下:

Attaches the given Sink to this Flow, meaning that elements that pass through this Flow will also be sent to the Sink.

对我来说这听起来像是广播，但也许这就是我犯错误的地方。我还可以解释为 toMat 控制流程。因此，我希望以下内容能够从源中获取所有元素:

Source(1 to 10)
  .alsoTo(Sink.head)
  .toMat(Sink.seq)(Keep.right)
  .run()

GraphDSL 版本如我所料:

val s1 = Sink.foreach[Int](v => println(s"each: $v"))
val s2 = Sink.head[Int]
val source = Source(1 to 10)
RunnableGraph.fromGraph(GraphDSL.create(s1, s2)((_, _)) { implicit builder => (s1, s2) =>
  import GraphDSL.Implicits._
  val broadcast = builder.add(Broadcast[Int](2))
  source ~> broadcast.in
  broadcast.out(0) ~> s1
  broadcast.out(1) ~> s2
  ClosedShape
}).run()

Answer 1

原因是 Sink.head 消耗单个元素并完成自身。这将以 cancel 的形式向上游传播，此后不会从 Source 发送任何元素。

akka.stream.impl.HeadOptionStage.onPush 中的代码展示了它

      def onPush(): Unit = {
        p.trySuccess(Option(grab(in)))
        completeStage()
      }

哪里 completeStage

Automatically invokes [[cancel()]] or [[complete()]] on all the input or output ports that have been called, then marks the operator as stopped.

更新

alsoTo 是使用以下参数配置的广播:

      val bcast = b.add(Broadcast[Out](2, eagerCancel = true))

您的 GraphDSL 版本工作方式不同，因为默认广播是 eagerCancel = false。

在哪里 eagerCancel

if true, broadcast cancels upstream if any of its downstreams cancel.