Scala Futures - 被两种方法的 CPU 负载和输出所迷惑

Question

我在实现 scala 期货时犯了一个错误，或者至少我认为我犯了，但是只是注意到它，当我修复错误时，它的运行速度比我不使用期货时慢得多。有人可以帮助我了解发生了什么吗？

我有一个慢速方法，需要运行 5,000 次。每个都是独立的并返回一个 Double。然后我需要计算 5,000 个返回值的平均值和标准差。

当我最初编码时，我是这样做的：

import actors.Futures._
import util.Random
import actors.Future

def one = {
  var results = List[Future[Double]]()
  var expectedResult: List[Double] = Nil
  var i = 0

  while (i < 1000) {
    val f = future {
      Thread.sleep(scala.util.Random.nextInt(5) * 100)
      println("Loop count: " + i)
      Random.nextDouble
    }
    results = results ::: List(f)
    println("Length of results list: " + results.length)

    results.foreach(future => {
      expectedResult = future() :: expectedResult
      i += 1
    })
  }
  // I would return the list of Doubles here to calculate mean and StDev
  println("### Length of final list: " + expectedResult.length)
}

我没有想到它，因为它运行得很快，而且我得到了预期的结果。当我仔细查看它以尝试让它运行得更快（它没有使用我所有可用的 CPU 资源）时，我意识到我的循环计数器在错误的位置，而 foreach 在里面future 创建循环，因此提前阻止了期货。或者我是这么想的。

我坚持了几个 println 语句，看看我是否能弄清楚发生了什么，并对发生的事情感到非常困惑......结果列表的长度与最终列表长度不匹配，也不匹配使用循环计数器！

我根据我认为（应该）发生的事情将我的代码修改为以下内容，并且事情变得慢得多，并且打印语句的输出与第一种方法相比没有任何意义。这次循环计数器似乎跳到了 1000，尽管最终的列表长度是有意义的。

第二种方法确实使用了所有可用的 CPU 资源，这更符合我的预期，但我确信结果相同需要更长的时间。

def two = {
  var results = List[Future[Double]]()
  var expectedResult: List[Double] = Nil
  var i = 0

  while (i < 1000) {
    val f = future {
      Thread.sleep(scala.util.Random.nextInt(5) * 100)
      println("Loop count: " + i)
      Random.nextDouble
    }
    results = f :: results
    i += 1
    println("Length of results list: " + results.length)

  }
  results.foreach(future => {
    expectedResult = future() :: expectedResult
  })
  // I would return the list of Doubles here to calculate mean and StDev
  println("### Length of final list: " + expectedResult.length)
}

我在这里遗漏了什么明显的东西吗？

---

编辑

对于任何看到这个的人......问题是我正在将期货的结果重新添加到期货循环中的最终列表 (expectedResult) - 正如 som-snytt 所指出的那样。

所以每次循环我都会反复迭代完成的期货并得到：

//First Loop: 
List(1)
//Second Loop:
List(1,2)
//Third Loop:
List(1,2,3,4)
//... and so on

最终列表中的模式是这样的：

List(n, n-1, n-2, ..., 4, 3, 2, 1, 3, 2, 1, 2, 1, 1)

由于列表的长度为 5050 项，并且值为 Double 值，因此当我只查看列表的开头时，很难看到模式。

最终循环数实际上只有 100 个，而不是我需要的 5000 个。

该方法的第二版对于 scala 2.9 是正确的。

Answer 1

我在这里遗漏了什么明显的东西吗？

没有。公平地说，命令式编程让一切变得不明显。

其中之一，您反复迭代结果，碰到i。

上次通过时间：

Length of results list: 45
Loop count: 990
### Length of final list: 1035

i 计算最终列表，应用 future 增加结果的长度，所以数学是正确的：45 + 990 = 1035。

应用完成的期货只是获得价值；您阻塞只是为了等待，因此您不一定会注意到一遍又一遍地获取未来值的性能问题。

但请注意，在未来，您将关闭 var i，请参阅Captured by Closures，而不是创建未来时 i 的值。令人困惑的是，由于缺乏同步，“循环计数”并不可靠。

我没有想到它，因为它跑得很快，我得到了 我预期的结果。

这个观察包含了很多工程智慧。

以下是 2.9 的另外两个公式：

  def four = (1 to 1000).par map { i =>
    Thread sleep nextInt(5) * 100
    Console println "Loop count: " + i
    nextDouble
  } 

  def three = 
    (1 to 1000) map (i => future {
        Thread sleep nextInt(5) * 100
        Console println "Loop count: " + i
        nextDouble
    }) map (_())

这是 2.10 中的新 API，仅供比较。

import scala.concurrent._
import scala.concurrent.duration._
import scala.util._

object Test extends App {
  import ExecutionContext.Implicits.global
  import Random._
  def compute(i: Int) = future {
    Thread.sleep(nextInt(5) * 100)
    val res = nextDouble
    println(s"#$i = $res")
    res
  }
  val f = Future.traverse(1 to 1000)(compute)
  val res = Await result (f, Duration.Inf)
  println(s"Done with ${res.length} results")
}