ConcurrentBag<T> 获取重复项(似乎不是线程安全的)

Question

我一定在哪里做错了什么，因为我的concurrentbag中有重复的项目，下面是事件链

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 source，Func`2谓词)处包含多个匹配元素
在d:\Projects\ListingLocator2\Code\LocalSearch.Processor\Processors.cs:line 310处
LocalSearch.Processor.CityProcessor.ProcessType(ConcurrentBag`1个列表，GeocodeResults状态，XElement类型，ParallelLoopState状态)中的CallSite.Target(闭包、CallSite、CityProcessor、Object、XElement、ConcurrentBag` 1)
(位于LocalSearch.PProcessor.CityProcessor.cs:line 249)。<>d:\Projects\ListingLocator2\Code\LocalSearch.Processor\Processors.cs:line 137 中的c__DisplayClass4.b_0(XElement类型，ParallelLoopState状态)

Answer 1

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

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

使用ConcurrentDictionary并利用TryAdd方法来更好地实现您的用例，是原子方法。或者，你可以在包周围放一个lock()，让块内的所有东西都原子化地运行，但这样你就不需要并发收集，而是可以使用直接的List。

Answer 2

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

您正在检查并发行李中的值，如果上次检查失败，则添加新项目。

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

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

你需要锁定你的支票并添加例程。