我需要并行处理多个数据值(“SIMD”)。我可以使用 java.util.concurrent
API (Executors.newFixedThreadPool()
) 使用 Future
实例并行处理多个值:
import java.util.concurrent.{Executors, Callable}
class ExecutorsTest {
private class Process(value: Int)
extends Callable[Int] {
def call(): Int = {
// Do some time-consuming task
value
}
}
val executorService = {
val threads = Runtime.getRuntime.availableProcessors
Executors.newFixedThreadPool(threads)
}
val processes = for (process <- 1 to 1000) yield new Process(process)
val futures = executorService.invokeAll(processes)
// Wait for futures
}
如何使用 Actors 做同样的事情?我不认为我想将所有流程“提供”给单个 Actor ,因为 Actor 将按顺序执行它们。
我是否需要创建多个“处理器”actor,并使用“调度程序”actor 向每个“处理器”actor 发送相同数量的进程?
最佳答案
如果您只想进行即发即弃处理,为什么不使用 Scala futures?
import scala.actors.Futures._
def example = {
val answers = (1 to 4).map(x => future {
Thread.sleep(x*1000)
println("Slept for "+x)
x
})
val t0 = System.nanoTime
awaitAll(1000000,answers: _*) // Number is timeout in ms
val t1 = System.nanoTime
printf("%.3f seconds elapsed\n",(t1-t0)*1e-9)
answers.map(_()).sum
}
scala> example
Slept for 1
Slept for 2
Slept for 3
Slept for 4
4.000 seconds elapsed
res1: Int = 10
基本上,您所做的就是将您想要的代码放入 future{}
block 中,它会立即返回一个 future;应用它来获得答案(它将阻塞直到完成),或者使用带有超时的 awaitAll
等待每个人都完成。
更新:从 2.11 开始,执行此操作的方法是使用 scala.concurrent.Future
。上面代码的翻译是:
import scala.concurrent._
import duration._
import ExecutionContext.Implicits.global
def example = {
val answers = Future.sequence(
(1 to 4).map(x => Future {
Thread.sleep(x*1000)
println("Slept for "+x)
x
})
)
val t0 = System.nanoTime
val completed = Await.result(answers, Duration(1000, SECONDS))
val t1 = System.nanoTime
printf("%.3f seconds elapsed\n",(t1-t0)*1e-9)
completed.sum
}
关于scala - 服务于一项任务的多个 Scala 参与者,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/5418439/