将 Iterator 映射到 Iterator[Future] 的行为不符合预期答案

【问题标题】：Mapping Iterator into Iterator[Future] is not behaving as expecting将 Iterator 映射到 Iterator[Future] 的行为不符合预期
【发布时间】：2021-02-04 17:58:45
【问题描述】：

使用 Scala 2.13.0：

    implicit val ec = ExecutionContext.global
    val arr = (0 until 20).toIterator
      .map { x =>
        Thread.sleep(500);
        println(x);
        x
      }

    val fss = arr.map { slowX =>
      Future { blocking { slowX } }
    }
    Await.result(Future.sequence(fss), Inf)

问题

arr 是一个迭代器，其中每个项目需要 500 毫秒的处理时间。我们用Future { blocking { ... }} 映射迭代器，目的是使处理并行（使用全局执行上下文）。最后我们运行Future.sequence 使用迭代器。

给定Future.apply[T](body: =>T)和blocking[T](body: =>T)的定义，body是惰性传递的，这意味着body将在Future中被处理。如果我们在Iterator.map的定义中注入它，我们得到def next() = Future{blocking(self.next())}，所以迭代器的每一项都应该在Future中处理。

但是当我尝试这个示例时，我可以看到迭代器是按顺序使用的，这不是预期的！

这是一个 Scala 错误吗？还是我错过了什么？

【问题讨论】：

arr.map 在传递给Future 之前计算前一个map 中的值，因此sleep 发生在Future 之外

标签： scala

【解决方案1】：

不，这不是错误，因为：

val arr = (0 until 20).toIterator
// this map invokes first and executed sequentially, because it executes in same thread.
      .map { x => 
        Thread.sleep(500);
        println(x);
        x
      }

    // This go sequentially because upstream map executed sequentially in same thread.
    // So, "Future { blocking { slowX } }" can be replaced with "Future.successfull(slowX)"
   // because no computation executed
    val fss = arr.map { slowX =>
      Future { blocking { slowX } }
    }

如果你想完全异步执行，你可以这样做：

def heavyCalculation(x: Int) = {
  Thread.sleep(500);
  println(x);
  x
}

val result = Future.traverse((0 until 20).toList) { x => 
  Future(blocking(heavyCalculation(x)))
}

Await.result(result, 1 minute)

工作Sctie示例：https://scastie.scala-lang.org/3v06NpypRHKYkqBgzaeVXg

【讨论】：

Iterator.map 本身没有任何动作：它只是定义输出迭代器。 sleep 仅在迭代器被消耗时调用（Future.sequence）。如果你关注Iterator.map的源代码，你会看到第一张地图的主体对应于next()，它在Future中运行。
@stackoverflowed 对不起，由于误解，我是在您的代码示例中领先，如果您可以，请提供您有问题的更精确的代码示例？如果你有 Iterator with slow next 方法，那么 hasNext 方法也很慢吗？这个迭代器线程安全吗？
@stackoverflowed 为什么你有一个Iterator 会阻止它的next？你能控制创建它的代码吗？
@LuisMiguelMejíaSuárez blocking 不会阻止！它通过告诉 ExecutionContext 做相反的事情：“这段代码可能在某些 IO 上阻塞，所以请在我等待时安排另一个任务”。您可以尝试在第一张地图中添加Future {...}，然后添加Future { blocking {...}}，您会看到明显的加速。
@LuisMiguelMejíaSuárez 现在一切都说得通了。感谢您的回答，因为他们让我思考并获得了更好的画面。

【解决方案2】：

首先，这不是一个合适的基准，实际上你还没有正式证明这是顺序的而不是并行的（尽管从源代码中“显而易见”它不是） .

其次，Iterator 的Futures 可能是个坏主意；在这一点上，研究像 Akka-Streams、fs2、Monix 或 ZIO 这样的流解决方案可能是有意义的强>。

第三，拥有一堆blocking 期货有什么意义？你实际上并没有赢得太多。

第四，问题是第二个map没有通过第一个map的块，只是结果。因此，您实际上是在创建 Future 之前执行了 sleep。

第五，你可能想要这样做。

val result = Future.traverse(data) { elem =>
  Future {
    blocking {
      // Process elem here.
    }
  }
}
Await.result(result, Inf)

【讨论】：

blocking 被精确使用，以便全局执行上下文不会阻塞；）请参阅此处docs.scala-lang.org/overviews/core/futures.html（全局执行上下文）
@stackoverflowed 这就是我的观点，你只是在创建一堆线程来阻止它们。重点是什么？不阻塞主线程？您在Await 中阻止它以等待所有阻止完成。异步启动所有阻塞操作？然后你必须小心，因为你最终可能会创建太多的线程，你会饿死你的记忆。
@stackoverflowed 另外，只有当迭代器能够并行产生多个元素时才会起作用，它可能不能，所以最后，它会比串行执行慢，因为线程的开销。最后，即使您不应该使用可变性+并发性，这只会导致难以调试的错误。使用适当的并发抽象（如Stream）。

【解决方案3】：

其他答案指向正确的方向，但正式的答案如下：Iterator.map(f: A => B) 的签名告诉我们 A A 是在 f 应用于它之前计算的（因为它不是=> A)。因此，next() 是在主线程中计算出来的。

【讨论】：