如何在 Parallel.ForEach 期间添加或更新 ConcurrentDictionary？

Question

我有一个文件列表，其中每个文件都包含一个Foo数据列表。现在，同一段 Foo 数据（例如Id = 1）可能存在于多个文件中，但最新的数据将覆盖现有的数据。

我只是将每条数据读入内存集合中。

if !cache.HasKey(foo.Id) then Add    
else cache[foo.Id].UpdatedOn < foo.UpdatedOn then Update  
else do nothing

当我阅读文件时（因为其中有一些），我也在使用Parallel.ForEach(files, file => { .. });

我不知道我该怎么做。

我正在考虑使用 aConcurrentDictionary但我不确定如何使用AddOrUpdatewhere子句。

有什么建议么？

Answer 1

您可以使用 a ConcurrentDictionary，如下所示：

dictionary.AddOrUpdate(foo.Id, foo, (id, existing) => 
    existing.UpdatedOn < foo.UpdatedOn ? foo : existing);

由于下面评论中的讨论，我将解释为什么这里没有竞争条件。这篇MSDN 文章讨论了价值工厂的运行方式，并提到：

因此，不能保证 GetOrAdd 返回的数据与线程的 valueFactory 创建的数据相同。

这是有道理的，因为并发字典的设计者不希望用户代码锁定字典（谁知道多久），从而使其变得无用。相反，它AddOrUpdate在两个嵌套循环中运行。这是一些伪代码：

do { 
   while (!TryGetValue(key, out value))
       if (TryAdd(key, addValue)) return;
   newValue = updateValueFactory(key, value);
} while (TryUpdate(key, newValue, value));

TryUpdate获取特定存储桶的锁，将当前值与检索到的值进行比较，只有当它们匹配时才执行更新。如果失败，外部循环会再次发生，TryGetValue返回最新值，再次调用值工厂，依此类推。

因此可以保证，如果更新成功，值工厂将始终具有最新值。