ConcurrentBag<T> 获取重复项（似乎不是线程安全的）答案

【问题标题】：ConcurrentBag<T> getting duplicates (seems not to be thread safe)ConcurrentBag<T> 获取重复项（似乎不是线程安全的）
【发布时间】：2014-02-07 12:02:35
【问题描述】：

我一定是在某处做错了，因为我的并发包中有重复的项目，这是事件链

  var listings = new ConcurrentBag<JSonListing>();
  Parallel.ForEach(Types, ParallelOptions, (type, state) =>
  {
      ...
      status = ProcessType(listings, status, type, state);
      ....
   });

  private GeocodeResults ProcessType(ConcurrentBag<JSonListing> listings, GeocodeResults status, XElement type, ParallelLoopState state)
  {
      ....
      AddListingsToList(results, type, listings);
      .... 
  }

private void AddListingsToList(dynamic results, XElement type, ConcurrentBag<JSonListing> listings)
    {

        var typeMustMatch = type.Attribute("TypeMustMatch").Value;
        var typeID = Convert.ToInt32(type.Attribute("ID").Value);

        foreach (var result in results.results)
        {


            var foundListing = listings.SingleOrDefault(x => x.ID == result.id);
            if (foundListing != null)
            {
                var typeIDs = foundListing.TypeIDs.Split("|".ToCharArray(), StringSplitOptions.RemoveEmptyEntries).ToList();
                if (!typeIDs.Contains(typeID.ToString()))
                {
                    foundListing.TypeIDs += "|" + typeID;
                }
            }
            else
            {
                var listing = new JSonListing { ID = result.id, ReferenceNumber = result.reference, TypeIDs = typeID.ToString(), TypeMustMatch = typeMustMatch };
                listings.Add(listing);
            }




        }


    }

添加列表应保证如果元素已经存在，则不会添加另一个相同 ID 的元素，而是更新某些属性。现在我得到的错误是

System.InvalidOperationException：序列包含多个匹配元素
在 System.Linq.Enumerable.SingleOrDefault[TSource]（IEnumerable`1 源，Func`2 谓词）
在 LocalSearch.Processor.CityProcessor.AddListingsToList（对象结果，XElement 类型，ConcurrentBag`1 列表）在 d:\Projects\ListingLocator2\Code\LocalSearch.Processor\Processors.cs:line 310
在 CallSite.Target(Closure , CallSite , CityProcessor , Object , XElement , ConcurrentBag`1 )
在 LocalSearch.Processor.CityProcessor.ProcessType（ConcurrentBag`1 列表，GeocodeResults 状态，XElement 类型，ParallelLoopState 状态）在 d:\Projects\ListingLocator2\Code\LocalSearch.Processor\Processors.cs:line 249
在 LocalSearch.Processor.CityProcessor.c__DisplayClass4.b__0(XElement type, ParallelLoopState state) in d:\Projects\ListingLocator2\Code\LocalSearch.Processor\Processors.cs:line 137

【问题讨论】：

使用 PLINQ 可以更干净地解决这个问题。
@spender：看起来ProcessType 会有显着的副作用，但示例中已省略。
@spender ProcessType 的唯一方法是调用（或多次调用）google places api，它返回与搜索查询匹配的可用位置的动态列表。 PLINQ 不适用于此，因为这些不在内存项中。

标签： c# parallel-extensions

【解决方案1】：

ConcurrentBag 保证其上的每个操作在单独考虑时都是线程安全的。它不保证连续的多个操作将被视为一个原子组。

因此，您的代码存在竞争条件：您检查包是否已经包含某个项目 X，但两个线程可以同时运行测试，确定该项目不存在，然后继续添加它。最终结果：该物品的两个副本最终在袋子中。

看起来您的用例可以通过使用ConcurrentDictionary 并利用TryAdd 方法来更好地实现，该方法是原子的。或者，您可以在包周围放置一个lock()，以使块内的所有内容都以原子方式运行，但是您实际上并不需要并发集合，而是可以直接使用List。

【讨论】：

我怎样才能做到没有重复进入袋子？
我认为最好是使用 List 让它有重复项，一旦完成通过该列表，这些列表中的大多数将有大约 4k 项，因此不会对性能造成重大影响（至少我不这么认为，但我会尝试的）
@Zoinky：你可以尝试很多东西，最好的取决于你的具体要求（我们只能猜测）。为什么不使用Set<T> 和适当的相等比较器，而不是允许欺骗？更好的是，既然这里有一个自然密钥可以使用，为什么不ConcurrentDictionary？
我相信 ConcurrentDictionary 将是我正在寻找的，我的要求非常简单。主要的 Parrallel.ForEach 有大约 20 个项目，每个项目都是对 google places API 的不同调用，现在这些类型中的每一个都将返回多达 200 个结果，类型 1 和类型 2 可能会返回相同的项目，但不会丢弃一个其中，我只是想通过将 typeid 附加到集合中已有的项目来跟踪它，所以如果 type1 和 type2 在带回相同的项目中，应该只有一个项目带有字符串“1 | 2”表示两者其中找到了该项目。

【解决方案2】：

这并不意味着 Concurrent Bag 不是线程安全的，它只是意味着您的代码不是

您正在检查并发包中的值，如果之前的检查失败，则添加一个新项目。

但是，由于您的代码中没有锁，因此两个作业可以同时执行以下操作；

THREAD 1        THREAD 2
=-=-=-=-=-=-=-=-=-=-=-=-
Check Exists
                Check Exists
Add New
                Add New

您需要锁定检查并添加例程。

【讨论】：