不完全理解Source.delay方法或akka-stream docs中有错误

Question

我正在阅读有关KillSwitch 的akka-stream 文档，他们有一个示例来说明该KillSwitch.shutdown方法：

val countingSrc = Source(Stream.from(1)).delay(1.second, DelayOverflowStrategy.backpressure)
val lastSnk = Sink.last[Int]

val (killSwitch, last) = countingSrc
  .viaMat(KillSwitches.single)(Keep.right)
  .toMat(lastSnk)(Keep.both)
  .run()

doSomethingElse()

killSwitch.shutdown()

Await.result(last, 1.second) shouldBe 2

我在理解为什么预期结果应该是 2 时遇到问题。正如我看到的这个例子，流被延迟了 1 秒。当它暂停时shutdown()被调用，因此终止开关告诉流在延迟完成之前关闭。我不明白为什么流的前 2 个元素会被发射并传送到接收器。

你能帮忙解释一下吗？

注意：如果我运行这个例子，我会得到我预期的以下异常：

Exception in thread "main" java.util.NoSuchElementException: last of empty stream
    at akka.stream.scaladsl.Sink$.$anonfun$last$3(Sink.scala:181)

Answer 1

对示例代码有误解。结果完全取决于 的运行时长doSomethingElse。只有当它花费的时间太少时，你才会得到例外。要对此进行测试，您可以将其替换为Thread.sleep(2000)，您将从Sink. 如果您增加睡眠值，结果也会增加。

关于您在评论中的问题：

delay按指定的量及时移动元素发射。延迟精度为 10ms，以避免不必要的定时器调度周期。这就是您看到此行为的原因（您可以在 的 Scala 文档中查看这些详细信息Flow）。

如果您想每秒发送一条消息，请尝试throttle：

.throttle(1, 1.second, 1, ThrottleMode.shaping)