如何结束无限的akka流答案

【问题标题】：How to end an infinite akka stream如何结束无限的akka流
【发布时间】：2015-11-01 12:55:14
【问题描述】：

我是 Akka Streams 的新手，但有一个案例我想用它来查找来自无限源的排列。具有有限源的简化示例可能如下所示。

val future = Source(1 to 100)
    .map { i => if (i % 20 == 0) println(i); i }
    .filter(_ == 42)
    .runWith(Sink.fold[Int, Int](0)(Keep.right))

这个例子输出：

我显然对源超过 42 没问题，但我不想在获得结果之前耗尽整个流。

val result: Int = Await.result(future, 1.second)
result should be(42)

问题是，当我找到我要找的内容后，我应该如何结束直播？

【问题讨论】：

标签： scala terminate akka-stream

【解决方案1】：

val future = Source(1 to 100)
    .map { i => if (i % 20 == 0) println(i); i }
    .filter(_ == 42)
    .runWith(Sink.head)

【讨论】：

【解决方案2】：

泛化为N 值，例如大于等于42的10个值，可以使用grouped：

val N = 10

val future : Future[Seq[Int]] = 
  Source(1 to 100).map { i => if (i % 20 == 0) println(i); i }
                  .filter(_ >= 42)
                  .grouped(N)
                  .runWith(Sink.head)

【讨论】：