Является ли этот код параллельного массива 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).

Кроме того, мне неясно, что дает вам параллельная коллекция, которая дает вам представление об изменении базовой коллекции во время обработки, поэтому буфер изменяемого списка, в который можно добавить элементы/ удален, возможно, плохой выбор.Вы никогда не знаете, когда следующий кодер вызовет что-то вроде foo(listBuffer) перед вашим foreach и передаст эту ссылку другому потоку, который может изменить список во время его обработки.

Кроме этого, я думаю, для любогоf, который займет много времени, может быть вызван одновременно, и когда e может быть обработан не по порядку, это хороший шаблон.

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

отказ от ответственности: я не пробовал// Собираюсь, но я с нетерпением жду, чтобы ТАК вопросы / ответы показали нам, что работает хорошо.

Answer 3

Код, который вы опубликовали, безопасен - не будет ошибок из-за несогласованного состояния вашего списка массивов, поскольку доступ к нему синхронизирован.

Однако параллельные коллекции обрабатывают элементы одновременно (в то же времявремя) И не в порядке.Превышение порядка означает, что элемент 54. может быть обработан до элемента 2. Ваш список синхронизированных массивов будет содержать элементы в неопределяемом порядке.

В общем случае лучше использовать map,filter и другие функциональные комбинаторы для преобразования коллекции в другую коллекцию - это обеспечит сохранение гарантий упорядочения, если в коллекции есть некоторые (как, например, Seq s).Например:

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

всегда возвращает ParArray(2, 3, 4, 5).

Однако, если вам нужен конкретный поточно-безопасный тип коллекции, такой как ConcurrentSkipListMap или синхронизированная коллекция, для передачи внекоторый метод в некотором API, модификация его из параллельного foreach безопасна.

Наконец, параллельный сбор примечаний обеспечивает параллельные массовые операции с данными.Изменяемые параллельные коллекции не являются поточно-ориентированными в том смысле, что вы можете добавлять к ним элементы из разных потоков.Изменяемые операции, такие как вставка на карту или добавление буфера, все еще должны быть синхронизированы.

Answer 4

Этот код довольно странный - зачем добавлять материал параллельно чему-то, что нужно синхронизировать? Вы добавите разногласия и ничего не получите взамен.

Принцип - накапливать результаты параллельной обработки лучше всего с такими вещами, как fold, reduce или aggregate.

Answer 5

synchronisedList должен быть безопасным, хотя println может дать неожиданные результаты - у вас нет гарантии того, что элементы будут напечатаны, или даже то, что ваши printlns не будут чередоваться в середине символа.

Синхронизированный список также вряд ли будет самым быстрым способом сделать это, более безопасное решение - map над неизменяемой коллекцией (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")

Вы также заметите, что этот код имеет примерно такую ​​же практическую полезность, как и ваш пример:)