scala - 不完全理解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 的运行时长.只有当它花费的时间太少时，你才会得到异常(exception)。要对此进行测试，您可以将其替换为 Thread.sleep(2000)你会从 Sink 得到一个结果.如果您增加 sleep 值，结果也会增加。

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

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

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