Parallel ForEach и ConcurrentBag

Question

У меня есть ConcurrentBag, подвергнутый операциям чтения/записи внутри Parallel.ForEach. В принципе, мне нужно проверить наличие предмета в сумке на основе нескольких свойств, и если нет совпадения, я добавляю его в сумку. Это действительно, очень медленно. Использование List<> без блокировки - это часть времени. Что случилось с этим кодом? Мне лучше использовать блокировку списка с помощью ReaderWriterLockSlim? Здесь я обрабатываю около 1 000 000 объектов.

var bag = new ConcurrentBag<Beneficiary>(); 

Parallel.ForEach(cx, _options, line => 
{ 
if (!bag.Any(o => 
     o.WinID == beneficiary.WinID && 
     o.ProductType == beneficiary.ProductType && 
     o.FirstName == beneficiary.FirstName && 
     o.LastName == beneficiary.LastName && 
     o.MiddleName == beneficiary.MiddleName)) 
{ 
     bag.Add(beneficiary);  
} 
} 

Answer 1

A ConcurrentBag<T> не оптимизирован для этого типа сценария. Он реализуется с использованием ThreadLocal<T>, что делает ваш конкретный случай использования медленным. Вы многократно повторяете всю коллекцию на многих потоках. Итерация по всей коллекции для проверки существования объекта также медленная.

Я предлагаю перегружать Beneficiary.GetHashCode и использовать ConcurrentDictionary<Beneficiary, byte>. Значение байта может быть просто проигнорировано, и это фактически одновременный hashset.

Answer 2

Итак, решение, которое у вас там, не является безопасным типом. Элементы могут быть добавлены в коллекцию во время повторного копирования через копию или даже после выполнения Any, но перед вызовом Add. Вы также выполняете линейный поиск, который не будет хорошо работать. Вам лучше использовать структуру на основе словаря, которая будет быстрее искать, и вам также необходимо убедиться, что весь метод, который у вас здесь, является логически атомарным.

Что вы можете сделать, это использовать ConcurrentDictionary и создать IEqualityComparer, который проверяет 5 свойств, которые вас волнуют, что позволит вам добавлять элементы в словарь при перезаписи дубликатов.

Конечно, это все имеет смысл только в том случае, если для создания каждого объекта требуется большая работа. Если все, что вы пытаетесь сделать, это собрать коллекцию отдельных элементов, тогда вполне вероятно, что попытка распараллеливать эту операцию просто не будет победой. Если это в основном все, что вы делаете, то ресурсы, которые каждый поток должен выполнять, будут в таком утверждении, что количество фактической распараллеливания, которое у вас будет, будет очень низким, почти наверняка будет меньше, чем накладные расходы на потоки будут стоить вам. Скорее всего, вам лучше будет использовать синхронный вызов Distinct.

Answer 3

Вы можете использовать ключ Tuple и использовать ConcurrentDictionary для хранения вашего объекта benificiary.

var dict = new ConcurrentDictionary<Tuple<int, object, string>, Beneficiary>(); 

Parallel.ForEach(cx, _options, line => 
{ 
    string fullname = string.Join("|", line.FirstName, line.LastName, line.MiddleName); 

    Tuple<int, object, string> key = new Tuple<int,object,string>(line.WinID, line.ProductType, fullname); 

    //if (!dict.ContainsKey(key)) optional line 
    { 
     dict.TryAdd(key, line);} 
    } 
}); 

После parallel.ForEach завершена, вы можете получить доступ к отчетливому получателю, используя простой ForEach.

Примечание: вы должны заменить тип «объект» на тип продукта ProductType.