scala - 使用返回 future 的函数遍历列表和流

Question

简介

Scala 的 Future ( new in 2.10 和 now 2.9.3 )是一个应用仿函数，这意味着如果我们有一个 traversable type F，我们可以将一个 F[A] 和一个函数 A => Future[B] 并将它们转换为一个 Future [F[B]]。

此操作在标准库中可用 Future.traverse 。 Scalaz 7还提供了一个更通用的遍历，如果我们从scalaz-contrib library导入Future的应用仿函数实例，我们可以在这里使用它。 .

这两个traverse方法在流的情况下表现不同。标准库遍历在返回之前消耗流，而 Scalaz 的 returns the future immediately :

import scala.concurrent._
import ExecutionContext.Implicits.global

// Hangs.
val standardRes = Future.traverse(Stream.from(1))(future(_))

// Returns immediately.
val scalazRes = Stream.from(1).traverse(future(_))

还有另一个区别，如 Leif Warner观察here 。标准库的 traverse 立即启动所有异步操作，而 Scalaz 则启动第一个操作，等待其完成，启动第二个操作，等待它，依此类推。

流的不同行为

通过编写一个函数，为流中的第一个值休眠几秒钟，可以很容易地显示第二个差异:

def howLong(i: Int) = if (i == 1) 10000 else 0

import scalaz._, Scalaz._
import scalaz.contrib.std._

def toFuture(i: Int)(implicit ec: ExecutionContext) = future {
  printf("Starting %d!\n", i)
  Thread.sleep(howLong(i))
  printf("Done %d!\n", i)
  i
}

现在 Future.traverse(Stream(1, 2))(toFuture) 将打印以下内容:

Starting 1!
Starting 2!
Done 2!
Done 1!

以及 Scalaz 版本 (Stream(1, 2).traverse(toFuture)):

Starting 1!
Done 1!
Starting 2!
Done 2!

这可能不是我们想要的。

对于列表呢？

奇怪的是，这两个遍历在列表上的行为在这方面是相同的 - Scalaz 不会等到一个 future 完成后再开始下一个。

另一个 future

Scalaz 还包括自己的 concurrent包及其自己的 future 实现。我们可以使用与上面相同的设置:

import scalaz.concurrent.{ Future => FutureZ, _ }

def toFutureZ(i: Int) = FutureZ {
  printf("Starting %d!\n", i)
  Thread.sleep(howLong(i))
  printf("Done %d!\n", i)
  i
}

然后我们得到了 Scalaz 在流列表以及流上的行为:

Starting 1!
Done 1!
Starting 2!
Done 2!

也许不那么令人惊讶的是，遍历无限流仍然会立即返回。

问题

此时我们确实需要一个表格来总结，但列表也可以:

• 具有标准库遍历的流:在返回之前消耗；不要等待每个 future 。
• 带有Scalaz遍历的Streams:立即返回；请等待每个 future 完成。
• 带有流的 Scalaz future:立即返回；请等待每个 future 完成。

还有:

• 包含标准库遍历的列表:无需等待。
• 使用 Scalaz 遍历列表:无需等待。
• 带有列表的 Scalaz future:请等待每个 future 完成。

这有什么意义吗？在列表和流上的此操作是否有“正确”的行为？是否有某种原因导致“最异步”行为(即，在返回之前不消耗集合，并且在继续下一个之前不等待每个 future 完成)未在此处表示？

Answer 1

我无法全部回答，但我尝试了一些部分:

Is there some reason that the "most asynchronous" behavior—i.e., don't consume the collection before returning, and don't wait for each future to complete before moving on to the next—isn't represented here?

如果您有相关计算且线程数量有限，则可能会遇到死锁。例如，您有两个 future 依赖于第三个 future(future 列表中的所有三个)，并且只有两个线程，您可能会遇到前两个 future 阻塞所有两个线程而第三个线程永远不会执行的情况。 (当然，如果你的池大小是1，即执行一次计算后执行另一次计算，你也会得到类似的情况)

为了解决这个问题，每个 future 都需要一个线程，没有任何限制。这适用于小型 future list ，但不适用于大型 future list 。因此，如果您并行运行所有程序，您将遇到这样一种情况:小示例在所有情况下都会运行，而较大的示例将陷入僵局。 (示例:开发人员测试运行良好，生产死锁)。

Is there a "correct" behavior for this operation on lists and streams?

我认为对于 future 来说这是不可能的。如果您了解更多的依赖关系，或者当您确定计算不会阻塞时，可能会出现更并发的解决方案。但执行 future list 对我来说是“被设计破坏的”。最好的解决方案似乎是一个，对于死锁的小例子来说，它已经失败了(即执行一个 Future 之后另一个)。

Scalaz futures with lists: do wait for each future to complete.

我认为 scalaz 在内部使用理解来进行遍历。对于推导式，不能保证计算是独立的。所以我猜想 Scalaz 在理解方面做了正确的事情:一个接一个地进行计算。对于 future，这将始终有效，因为您的操作系统中有无限的线程。

换句话说:您看到的只是理解式(必须)如何工作的产物。

我希望这是有道理的。