这个scala并行数组代码线程安全吗？

Question

我想使用并行数组来完成任务，在开始编码之前，我有兴趣知道这个小片段是否是线程安全的：

import collection.mutable._

var listBuffer = ListBuffer[String]("one","two","three","four","five","six","seven","eight","nine")
var jSyncList  = java.util.Collections.synchronizedList(new java.util.ArrayList[String]())
listBuffer.par.foreach { e =>
    println("processed :"+e)
    // using sleep here to simulate a random delay
    Thread.sleep((scala.math.random * 1000).toLong)
    jSyncList.add(e)
}
jSyncList.toArray.foreach(println)

是否有更好的方法来处理并行集合的某些内容，并在其他地方累积结果？

Answer 1

您发布的代码是完全安全的；但我不确定这个前提：为什么需要在非并行集合中累积并行集合的结果？并行集合的要点之一是它们看起来像其他集合。

我认为并行集合还将提供一个 seq 方法来切换到顺序集合。所以你应该使用这个！

Answer 2

为了使此模式安全：

listBuffer.par.foreach { e => f(e) }

f 必须能够以安全的方式同时运行。我认为安全多线程所需的相同规则也适用（对共享状态的访问需要是线程安全的，f 调用不同 e 的顺序不会'不要具有确定性，当您开始同步 f 中的语句时，您可能会遇到死锁。

此外，我不清楚什么可以保证并行集合在处理时为您提供有关底层集合被修改的保证，因此可以添加/删除元素的可变列表缓冲区可能是一个糟糕的选择。您永远不知道下一个编码器何时会在 foreach 之前调用类似 foo(listBuffer) 的内容，并将该引用传递给另一个线程，这可能会在处理列表时改变列表。

除此之外，我认为对于任何需要很长时间、可以并发调用并且可以乱序处理的 f 来说，这是一个很好的模式。

immutCol.par.foreach { e => threadSafeOutOfOrderProcessingOf(e) }

免责声明：我自己没有尝试过// colls，但我期待着有这样的问题/答案向我们展示什么是有效的。

Answer 3

synchronizedList 应该是安全的，尽管 println 可能会给出意想不到的结果 - 您无法保证项目将被打印的顺序，甚至您的 println 也不会被打印。字符中间交错。

同步列表也不太可能是执行此操作的最快方法，更安全的解决方案是在不可变集合上映射（Vector 可能是您最好的选择），然后打印所有行（按顺序）：

val input = Vector("one","two","three","four","five","six","seven","eight","nine")
val output  = input.par.map { e =>
  val msg = "processed :" + e
  // using sleep here to simulate a random delay
  Thread.sleep((math.random * 1000).toLong)
  msg
}
println(output mkString "\n")

您还会注意到此代码与您的示例具有同样多的实际用途:)

Answer 4

这段代码很奇怪——为什么要并行地添加一些需要同步的东西呢？您将增加争用，但绝对不会获得任何回报。

事情的原理 - 累积并行处理的结果，可以通过fold、reduce或aggregate等东西更好地实现代码>.

Answer 5

您发布的代码是安全的 - 不会由于数组列表的状态不一致而出现错误，因为对它的访问是同步的。

但是，并行集合会同时（同时）且无序地处理项目。无序意味着 54. 元素可能会在 2. 元素之前处理 - 您的同步数组列表将包含非预定义顺序的项目。

一般来说，最好使用 map、filter 和其他函数组合器将一个集合转换为另一个集合 - 如果集合有一些 (就像 Seq 所做的那样）。例如：

ParArray(1, 2, 3, 4).map(_ + 1)

始终返回 ParArray(2, 3, 4, 5)。

但是，如果您需要将特定的线程安全集合类型（例如 ConcurrentSkipListMap 或同步集合）传递给某些 API 中的某些方法，则从并行 foreach 修改它是安全的。

最后，注意 - 并行集合提供对数据的并行批量操作。可变并行集合不是线程安全的，因为您可以从不同线程向它们添加元素。诸如插入映射或附加缓冲区之类的可变操作仍然必须同步。