C# 和 ConcurrentDictionary 中的多线程：以下用法正确吗？

Question

我手头有这样一个场景（使用 C#）：我需要在对象列表上使用并行“foreach”：此列表中的每个对象都像数据源一样工作，它正在生成一系列二进制向量模式（如“0010100110”）。在生成每个向量模式时，我需要更新共享 ConcurrentDictionary 上当前向量模式的出现次数。这个 ConcurrentDictionary 就像所有数据源中特定二进制模式的直方图。在伪代码中它应该像这样工作：

ConcurrentDictionary<BinaryPattern,int> concDict = new ConcurrentDictionary<BinaryPattern,int>();
Parallel.Foreach(var dataSource in listOfDataSources)
{
     for(int i=0;i<dataSource.OperationCount;i++)
     {
          BinaryPattern pattern = dataSource.GeneratePattern(i);

          //Add the pattern to concDict if it does not exist, 
          //or increment the current value of it, in a thread-safe fashion among all
          //dataSource objects in parallel steps.
     }
}

我在文档中阅读了 ConcurrentDictionary 类的 TryAdd() 和 TryUpdate() 方法，但我不确定我是否清楚地理解了它们。 TryAdd() 获取当前线程对 Dictionary 的访问权，并查找特定键是否存在，在这种情况下为二进制模式，然后如果不存在，则创建其条目，将其值设置为 1是这种模式的第一次出现。 TryUpdate() 获得对当前线程字典的访问权，查看具有指定键的条目的当前值是否等于“已知”值，如果是，则更新它。顺便说一句，TryGetValue() 检查字典中是否存在某个键，如果存在则返回当前值。

现在我想到以下用法，想知道它是否是 ConcurrentDictionary 的线程安全填充的正确实现：

ConcurrentDictionary<BinaryPattern,int> concDict = new ConcurrentDictionary<BinaryPattern,int>();
Parallel.Foreach(var dataSource in listOfDataSources)
{
     for(int i=0;i<dataSource.OperationCount;i++)
     {
          BinaryPattern pattern = dataSource.GeneratePattern(i);

          while(true)
          {
             //Look whether the pattern is in dictionary currently,
             //if it is, get its current value.
             int currOccurenceOfPattern;
             bool isPatternInDict = concDict.TryGetValue(pattern,out currOccurenceOfPattern);

             //Not in dict, try to add.
             if(!isPatternInDict)
             {
                  //If the pattern is not added in the meanwhile, add it to the dict.
                  //If added, then exit from the while loop.
                  //If not added, then skip this step and try updating again.
                  if(TryAdd(pattern,1))
                        break;
             }
             //The pattern is already in the dictionary. 
             //Try to increment its current occurrence value instead.
             else
             {
                  //If the pattern's occurence value is not incremented by another thread
                  //in the meanwhile, update it. If this succeeds, then exit from the loop.
                  //If TryUpdate fails, then we see that the value has been updated
                  //by another thread in the meanwhile, we need to try our chances in the next
                  //step of the while loop.                   
                  int newValue = currOccurenceOfPattern + 1;
                  if(TryUpdate(pattern,newValue,currOccurenceOfPattern))
                       break;
             }

          }
     }
}

我试图将我的逻辑牢牢地总结在 cmets 中的上述代码 sn-p 中。根据我从文档中收集到的信息，考虑到 ConcurrentDictionary 的原子“TryXXX()”方法，可以以这种方式对线程安全的更新方案进行编码。这是解决问题的正确方法吗？如果不是，如何改进或更正？

Answer 1

首先，这个问题有点令人困惑，因为不清楚您所说的Parallel.Foreach 是什么意思。我天真地认为这是System.Threading.Tasks.Parallel.ForEach()，但这不适用于您在此处显示的语法。

也就是说，假设您实际上是指 Parallel.ForEach(listOfDataSources, dataSource => { ... } )...

就个人而言，除非您有特定的需要显示中间结果，否则我不会在这里打扰ConcurrentDictionary。相反，我会让每个并发操作生成自己的计数字典，然后在最后合并结果。像这样的：

var results = listOfDataSources.Select(dataSource =>
    Tuple.Create(dataSource, new Dictionary<BinaryPattern, int>())).ToList();

Parallel.ForEach(results, result =>
{
    for(int i = 0; i < result.Item1.OperationCount; i++)
    {
        BinaryPattern pattern = result.Item1.GeneratePattern(i);
        int count;

        result.Item2.TryGetValue(pattern, out count);
        result.Item2[pattern] = count + 1;
    }
});

var finalResult = new Dictionary<BinaryPattern, int>();

foreach (result in results)
{
    foreach (var kvp in result.Item2)
    {
        int count;

        finalResult.TryGetValue(kvp.Key, out count);
        finalResult[kvp.Key] = count + kvp.Value;
    }
}

这种方法将避免工作线程之间的争用（至少在涉及计数的情况下），从而可能提高效率。最终的聚合操作应该非常快，并且可以在单个原始线程中轻松处理。

Answer 2

我不知道BinaryPattern 在这里是什么，但我可能会以不同的方式解决这个问题。不要像这样复制值类型，将内容插入字典等，如果性能很关键，只需将实例计数器放在BinaryPattern 中，我可能会更倾向于。然后在找到该模式时使用InterlockedIncrement() 递增计数器。

除非有理由将计数与模式分开，在这种情况下，ConccurentDictionary 可能是一个不错的选择。

Answer 3

您可以使用AddOrUpdate 方法将添加或更新逻辑封装为单个线程安全操作：

ConcurrentDictionary<BinaryPattern,int> concDict = new ConcurrentDictionary<BinaryPattern,int>();
Parallel.Foreach(listOfDataSources, dataSource =>
{
    for(int i=0;i<dataSource.OperationCount;i++)
    {
        BinaryPattern pattern = dataSource.GeneratePattern(i);

        concDict.AddOrUpdate(
            pattern,
            _ => 1, // if pattern doesn't exist - add with value "1"
            (_, previous) => previous + 1 // if pattern exists - increment existing value
        );
    }
});

请注意AddOrUpdateoperation 不是原子的，不确定这是否是您的要求，但如果您需要知道将值添加到字典时的确切迭代，您可以保留您的代码（或将其提取到某种扩展方法）

您可能还想通过this article