Перечислите<T> поток безопасности

Question

Я использую код ниже

var processed = new List<Guid>();
Parallel.ForEach(items, item => 
{
    processed.Add(SomeProcessingFunc(item));
});

Является ли приведенный выше поток кода безопасным? Существует ли вероятность повреждения обработанного списка? Или я должен использовать блокировку перед добавлением?

var processed = new List<Guid>();
Parallel.ForEach(items, item => 
{
    lock(items.SyncRoot)
        processed.Add(SomeProcessingFunc(item));
});

Благодарю.

Answer 1

Нет! Это совсем не безопасно, потому что processed.Add не является. Вы можете сделать следующее:

items.AsParallel().Select(item => SomeProcessingFunc(item)).ToList();

Имейте в виду, что Parallel.ForEach был создан в основном для императивных операций для каждого элемента последовательности. То, что вы делаете, это карта: проецируйте каждое значение последовательности. Что это Select был создан для. AsParallel масштабирует его между потоками наиболее эффективным способом.

Этот код работает правильно:

var processed = new List<Guid>();
Parallel.ForEach(items, item => 
{
    lock(items.SyncRoot)
        processed.Add(SomeProcessingFunc(item));
});

но не имеет смысла с точки зрения многопоточности. lockПоскольку на каждой итерации выполняется полностью последовательное выполнение, группа потоков будет ожидать один поток.

Answer 2

Использование:

var processed = new ConcurrentBag<Guid>();

Смотрите параллельный цикл foreach - странное поведение.

Answer 3

Цитирую Джона Скита, прежде чем он доберется сюда:

В рамках Parellel Extensions в.Net 4, в новой System.Collections.Concurrent Пространство имен. Они предназначены для обеспечения безопасности при одновременных операциях из нескольких потоков с относительно небольшой блокировкой.

Они включают IProducerConsumerCollection<T>, BlockingCollection<T>, ConcurrentBag<T>, ConcurrentQueue<T>, ConcurrentStack<T>, and ConcurrentDictionary<TKey, TValue> среди других.

Answer 4

Использование ConcurrentBag типа Something

var bag = new ConcurrentBag<List<Something>>;
var items = GetAllItemsINeed();
Parallel.For(items,i =>                          
   {
      bag.Add(i.DoSomethingInEachI());
   });

Answer 5

Как альтернатива ответу Андрея:

items.AsParallel().Select(item => SomeProcessingFunc(item)).ToList();

Вы могли бы также написать

items.AsParallel().ForAll(item => SomeProcessingFunc(item));

Это делает запрос, стоящий за ним, еще более эффективным, потому что MSDN не требует слияния. Убедитесь, что SomeProcessingFunc функция потокобезопасна. И я думаю, но не проверял это, что вам все еще нужна блокировка, если список может быть изменен в другом потоке (добавление или удаление) элементов.

Answer 6

Чтение потокобезопасно, а добавление - нет. Вам нужна настройка блокировки чтения / записи, поскольку добавление может привести к изменению размера внутреннего массива, что может привести к неправильному чтению.

Если вы можете гарантировать, что размер массива не будет изменяться при добавлении, вы можете безопасно добавлять его во время чтения, но не указывайте мне это.

Но на самом деле список - это просто интерфейс к массиву.