如何在scala中并行迭代地图？

Question

我有一个如下代码.

var arr = new Array[...](map.size)
var i = 0
for((k,value) <- map)
{
    arr(i) = (k, someFunc(value))
    i += 1
}

我希望这个循环并行执行.例如,我希望它并行运行在8个独立的线程中.我怎样才能在Scala中实现这一目标？

Answer 1

您可以将映射转换为并行集合,并通过使用大小为8(或任何您想要的大小)的池覆盖默认的"TaskSupport"来控制线程数:

import scala.collection.parallel.ForkJoinTaskSupport
import scala.collection.parallel.immutable.ParMap

val parMap: ParMap[Int, Int] = map.par
parMap.tasksupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(8))

parMap.foreach { case (k, value) =>
  arr(i) = (k, someFunc(value))
  i += 1
}

请注意,您可以通过删除所有可变值来使此代码更"惯用" :

val arr = parMap.map { case (k, value) => (k, someFunc(value)) }.toArray
val i = arr.length

编辑:或者,更短的版本:

val arr = parMap.mapValues(someFunc).toArray
val i = arr.length