multithreading - 如何在 Scala 中并行迭代 map ？

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