并发收集支持删除指定的项目？

Question

非常简单:除了ConcurrentDictionary(我将使用它,但它不是真正正确的概念),是否有任何Concurrent集合(IProducerConsumer实现)支持基于项或谓词的简单相等删除特定项定义删除条件？

说明:我有一个多线程,多阶段的工作流算法,它从数据库中提取对象并将它们放在"起始"队列中.从那里他们被下一阶段抓住,进一步努力,并塞进其他队列.这个过程将持续几个阶段.同时,第一个阶段由其主管再次调用,并将对象拉出数据库,这些对象可以包括仍处于进行中的对象(因为它们尚未完成处理,因此没有使用标记集重新保留他们已经完成了).

我正在设计的解决方案是一个主要的"工作"集合; 当第一阶段检索到对象进行处理时,对象进入该队列,并且在完成必要处理的工作流程的任何阶段"处理"后,对象被重新保存到数据库后被删除.当对象在该列表中时,如果第一阶段重新检索它,它将被忽略.

我曾计划使用ConcurrentBag,但唯一的删除方法(TryTake)从包中删除任意项,而不是指定的项(并且.NET 4中的ConcurrentBag很慢).ConcurrentQueue和ConcurrentStack也不允许删除除了它将给你的下一个项目之外的项目,留下ConcurrentDictionary,它可以工作但是比我需要的更多(我真正需要的是存储正在处理的记录的Id;它们在工作流程中不会改变).

Answer 1

没有这种数据结构的原因是所有集合都具有查找操作时间O(n).这些IndexOf,Remove(element)等他们都通过枚举所有元素,并检查它们是否相等.

只有散列表的查找时间为O(1).在并发场景中,O(n)查找时间将导致集合的锁定非常长.其他线程在此期间将无法添加元素.

在字典中,只有被哈希命中的单元格才会被锁定.当通过散列单元格中的元素检查相等性时,其他线程可以继续添加.

我的建议是继续使用ConcurrentDictionary.

顺便说一下,对于你的解决方案来说,ConcurrentDictionary有点过大了.你真正需要的是快速检查物体是否正在工作.A HashSet将是一个完美的.它基本上什么都不做,然后Add(element),Contains(element),Remove(element).ConcurrentHeshSetjava中有一个实现.对于c#我发现了这个:如何在.Net中实现ConcurrentHashSet,不知道它有多好.

作为第一步,我仍然会编写一个带有HashSet接口的包装器ConcurrentDictionary来启动并运行,然后尝试不同的实现并查看性能差异.

Answer 2

正如已经解释过的其他帖子不可能从a Queue或ConcurrentQueue默认情况下删除项目,但实际上最简单的方法是扩展或包装项目.

public class QueueItem
{
    public Boolean IsRemoved { get; private set; }
    public void Remove() { IsRemoved = true; }
}

出列时:

QueueItem item = _Queue.Dequeue(); // Or TryDequeue if you use a concurrent dictionary
if (!item.IsRemoved)
{
    // Do work here
}