时间戳

首页 > 解决方案 > Akka 流：为什么 Sink.head 用alsoTo 广播终止流？

scala - Akka 流：为什么 Sink.head 用alsoTo 广播终止流？

问题描述

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

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说明如下：

将给定的 Sink 附加到此 Flow，这意味着通过此 Flow 的元素也将被发送到 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()

标签： scalaakka-stream

解决方案

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

来自的代码akka.stream.impl.HeadOptionStage.onPush显示它

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

在哪里completeStage

在所有已调用的输入或输出端口上自动调用 [[cancel()]] 或 [[complete()]]，然后将操作符标记为已停止。

更新

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

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

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

在哪里eagerCancel

如果为真，则广播在其任何下游取消时取消上游。

推荐阅读