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

Question

简而言之：除了ConcurrentDictionary(如果必须我会使用它，但这不是正确的概念)，是否有任何支持移除特定项目的并发集合(IProducerConsumer实现)，其基于一个项目的简单相等性或定义一个条件用于移除的谓词?

解释：我有一个多线程、多阶段的工作流算法，从数据库中提取对象并将它们放入“启动”队列。然后它们被下一个阶段抓取，进一步处理，并被塞入其他队列。这个过程在几个阶段中继续进行。同时，第一阶段由其监管者再次调用并从数据库中提取对象，这些对象可能包括仍在处理过程中的对象(因为它们还没有完成处理，所以还没有重新保存并设置标志，表明它们已经完成)。

我设计的解决方案是一个主要的“正在处理中”收集;当对象被检索出来由第一阶段处理时，它们进入该队列，并在它们被处理所需的工作流阶段完成之后被移除，重新保存到数据库中作为“已处理”。在对象在列表中的时候，如果被第一阶段重新检索，它将被忽略。

我原计划使用ConcurrentBag，但唯一的移除方法(TryTake)从袋子里移除任意一个项目，而不是指定的一个(并且ConcurrentBag在.NET 4中很慢)。ConcurrentQueue和ConcurrentStack也不允许移除项，除了下一个将要给你的，只留下ConcurrentDictionary，这个会起作用，但我不需要这么多(我真正需要的只是存储正在处理的记录的Id;它们在工作流过程中不会改变)。

- KeithS

---

你对使用ReaderWriterLockSlim和List的感觉如何？或者也许你可以自己编写并发集合。 - Frobzig

@Frobzig - 持保留态度到稍微感兴趣。我喜欢并发集合，因为它们只需要工作；几乎没有涉及到代码。 - KeithS

类似 Kafka 这样的工具非常适合处理队列，而不是试着自己编写。 - Mark Homer

Answer 1

没有这样的数据结构的原因是所有集合都具有O(n)的查找操作时间。这些包括IndexOf、Remove(element)等操作，它们枚举所有元素并检查它们是否相等。

只有哈希表具有O(1)的查找时间。在并发场景中，O(n)的查找时间将导致集合长时间锁定。在此期间，其他线程将无法添加元素。

在字典中，只有由哈希命中的单元格会被锁定。在一个线程通过哈希单元格中的元素检查相等性时，其他线程可以继续添加元素。

我的建议是继续使用ConcurrentDictionary。

顺便说一下，你说得对，ConcurrentDictionary 对于你的解决方案来说有点过大了。你真正需要的是快速检查一个对象是否正在运行。 HashSet 是完美的选择，它基本上仅执行Add(element)、Contains(element)、 Remove(element) 操作。 Java 中有一个ConcurrentHeshSet实现。对于C#，我在这里找到了一个实现：如何在.Net中实现ConcurrentHashSet，但我不知道它的好坏。

作为第一步，我仍会编写一个具有HashSet接口的包装器，将其启动并运行，然后尝试不同的实现，并查看性能差异。

- George Mamaladze

Answer 2

正如其它帖子已经解释过的那样，无法默认删除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
}

- Felix K.

---

这个解决方案会导致内存泄漏。 - Kirill Bestemyanov

@KirillBestemyanov 不会，它怎么可能引起内存泄漏呢？ - Felix K.

你没有从集合中删除任务，因此由于其运行而导致该集合消耗的内存仅增加了。 - Kirill Bestemyanov

@KirillBestemyanov 如果你想在移除项目时将它们删除，那么你不应该使用队列，而是使用其他东西。根据定义，这不是内存泄漏，而是对队列的误用。 - Felix K.

你为什么向提问者推荐这个解决方案？如果这是误用的话，应用程序在执行过程中会慢慢消耗内存。 - Kirill Bestemyanov

Answer 3

在通用意义上使集合线程安全真的很难。有太多因素影响着它是否能真正做到“线程安全”，而这些因素超出了库/框架类的责任或视野范围...正如你所指出的，其中一个缺点是性能。不可能编写既具有良好性能又线程安全的集合，因为它必须假定最坏情况...

通常推荐的做法是使用任何你想要的集合，并以线程安全的方式访问它。这基本上就是为什么框架中没有更多线程安全集合的原因。更多信息可以在http://blogs.msdn.com/b/bclteam/archive/2005/03/15/396399.aspx#9534371找到。

- Peter Ritchie

---

线程安全访问集合和访问线程安全集合之间的区别可能非常大。如果线程数很多，它们执行的操作又比较复杂，那么每个线程都在等待自己的锁并清除它，可能会导致巨大瓶颈。我想你可以维护一个任意大的锁对象集合（例如对应于每个哈希），但那将非常混乱... - Patrick M

@PatrickM：将特定应用程序访问集合变为线程安全的操作集，比在每种情况下都使集合线程安全要少得多。你只需要一个锁定对象即可使任何给定对象（无论是集合还是其他）的访问变为线程安全。所有Concurrent*集合都只使用一个锁定对象。 - Peter Ritchie

在许多情况下，设计快速线程安全的集合并不难，这些集合实现了某些常规接口的子集（例如，支持Add和按索引写入作为唯一变异方法的IList<T>）。这样的东西可能比提供更广泛访问的集合更有效，但在框架中很少见。我能想到的唯一一个例子是ConditionalWeakTable，它类似于字典，但不允许删除项。 - supercat