对期货作出反应

Question

我正在尝试使用分而治之（又名分叉/连接）方法来解决数字运算问题。这是代码：

import scala.actors.Futures.future

private def compute( input: Input ):Result = {
  if( pairs.size < SIZE_LIMIT ) {
    computeSequential()
  } else {
    val (input1,input2) = input.split
    val f1 = future( compute(input1) )
    val f2 = future( compute(input2) )
    val result1 = f1()
    val result2 = f2()
    merge(result1,result2)
  }
}

它运行（具有很好的加速），但是未来的 apply 方法似乎会阻塞线程并且线程池极大地增加。当创建太多线程时，计算就会被卡住。

是否有一种用于释放线程的 future 的 react 方法？或者任何其他方式来实现这种行为？

编辑：我正在使用 scala 2.8.0.final

Answer 1

不要声明（应用）您的 Future，因为这会迫使他们阻塞并等待答案；正如您所看到的，这可能会导致僵局。相反，使用它们来告诉他们完成后要做什么。而不是：

val result1 = f1()
val result2 = f2()
merge(result1,result2)

试试这个：

for {
  result1 <- f1
  result2 <- f2
} yield merge(result1, result2)

结果将是一个包含合并结果的 Responder[Result] （本质上是一个 Future[Result]）；您可以使用 respond() 或 foreach() 对此最终值执行一些有效的操作，也可以使用 map() 或 flatMap () 将其发送给另一个 Responder[T]。无需阻塞，只需为未来安排计算即可！

编辑 1：

好的，compute 函数的签名现在必须更改为 Responder[Result]，那么这对递归调用有何影响？让我们试试这个：

private def compute( input: Input ):Responder[Result] = {
  if( pairs.size < SIZE_LIMIT ) {
    future(computeSequential())
  } else {
    val (input1,input2) = input.split
    for {
      result1 <- compute(input1)
      result2 <- compute(input2)
    } yield merge(result1, result2)
  }
}

现在您不再需要使用 future(...) 包装对 compute 的调用，因为它们已经返回 Responder （Future 的超类）。

编辑 2：

使用这种连续传递样式的一个好处是，您的顶级代码（无论最初调用 compute 的内容）都不再阻塞。如果它是从 main() 调用的，而这就是程序所做的所有事情，这将是一个问题，因为现在它只会产生一堆 future，然后立即关闭，完成所有事情告诉去做。您需要做的是阻止所有这些未来，但仅在顶层一次，并且仅针对所有计算的结果，而不是任何中间计算的结果。

不幸的是，由 compute() 返回的 Responder 不再像 Future 那样具有阻塞 apply() 方法代码> 做了。我不确定为什么 flatMapping Futures 会生成通用的 Responder 而不是 Future；这似乎是一个 API 错误。但无论如何，您应该能够自己创建：

def claim[A](r:Responder[A]):A = {
  import java.util.concurrent.ArrayBlockingQueue
  import scala.actors.Actor.actor

  val q = new ArrayBlockingQueue[A](1)
  // uses of 'respond' need to be wrapped in an actor or future block
  actor { r.respond(a => q.put(a)) } 
  return q.take
}

所以现在您可以在 main 方法中创建一个阻塞调用来计算，如下所示：

val finalResult = claim(compute(input))