我试图使用分而治之(又名叉/连接)的方法来解决一个数字处理问题。以下是代码:
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)
}
}
它运行(有一个很好的速度),但未来的应用方法似乎阻塞一个线程和线程池大幅度增加。当过多的线程被创建时,计算就会被堆积。
期货是否有一种释放线的反应方法?或者其他方式来达到这种行为?
编辑:我正在使用Scala2.8.0。
发布于 2010-10-04 08:40:04
不要声称(应用)您的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)
}
}
现在,您不再需要将对compute
的调用包装为future(...)
,因为它们已经返回Responder
(Future
的一个超类)。
编辑2:
使用这种延续传递样式的一个结果是,您的顶层代码--不管最初叫什么compute
--都不再阻塞。如果它是从main()
调用的,而这是程序所做的全部,这将是一个问题,因为现在它只会产生一堆未来,然后立即关闭,完成了所有被要求做的事情。你需要做的是对所有这些未来进行block
,但只有一次,在最高级,并且只针对所有计算的结果,而不是任何中间计算结果。
不幸的是,由Responder
返回的compute()
不再像Future
那样具有阻塞的apply()
方法。我不知道为什么flatMapping Future
s会生成一个通用的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))
https://stackoverflow.com/questions/3851742
复制相似问题