如何访问阻塞集合的底层默认并发队列？

Question

我有多个生产者和一个消费者。但是，如果队列中有尚未消耗的内容，生产者不应再次将其排队。 （使用默认并发队列的唯一无重复阻塞集合）

if (!myBlockingColl.Contains(item))
    myBlockingColl.Add(item)

但是阻塞集合没有 contains 方法，也不提供任何类似 TryPeek() 的方法。如何访问底层并发队列，以便我可以执行类似

if (!myBlockingColl.myConcurQ.trypeek(item)
  myBlockingColl.Add(item)

In a tail spin 的操作？

Answer 1

这是一个有趣的问题。这是我第一次看到有人要求一个忽略重复的阻塞队列。奇怪的是，我在 BCL 中找不到任何像你想要的东西。我说这很奇怪，因为 BlockingCollection 可以接受 IProducerConsumerCollection 作为底层集合，该集合具有 TryAdd 方法，该方法被宣传为能够在以下情况下失败检测到重复项。问题是我没有看到 IProducerConsumerCollection 的具体实现来防止重复。至少我们可以自己写。

public class NoDuplicatesConcurrentQueue<T> : IProducerConsumerCollection<T>
{
  // TODO: You will need to fully implement IProducerConsumerCollection.

  private Queue<T> queue = new Queue<T>();

  public bool TryAdd(T item)
  {
    lock (queue)
    {
      if (!queue.Contains(item))
      {
        queue.Enqueue(item);
        return true;
      }
      return false;
    }
  }

  public bool TryTake(out T item)
  {
    lock (queue)
    {
      item = null;
      if (queue.Count > 0)
      {
        item = queue.Dequeue();
      }
      return item != null;
    }
  }
}

现在我们有了不接受重复项的 IProducerConsumerCollection，我们可以像这样使用它：

public class Example
{
  private BlockingCollection<object> queue = new BlockingCollection<object>(new NoDuplicatesConcurrentQueue<object>());

  public Example()
  {
    new Thread(Consume).Start();
  }

  public void Produce(object item)
  {
    bool unique = queue.TryAdd(item);
  }

  private void Consume()
  {
    while (true)
    {
      object item = queue.Take();
    }
  }
}

您可能不喜欢我的 NoDuplicatesConcurrentQueue 实现。如果您认为需要 TPL 集合提供的低锁性能，您当然可以自由地使用 ConcurrentQueue 或其他任何方式来实现。

更新：

今天早上我能够测试代码。有一些好消息和坏消息。好消息是这在技术上是可行的。坏消息是您可能不想这样做，因为 BlockingCollection.TryAdd 会拦截底层 IProducerConsumerCollection.TryAdd 方法的返回值，并在 检测到 false。是的，没错。它不会像您期望的那样返回 false ，而是生成异常。我必须说实话，这既令人惊讶又荒谬。 TryXXX 方法的重点是它们不应该抛出异常。我深感失望。

Answer 2

除了 Update 之后 Brian Gideon 提到的警告之外，他的解决方案还存在以下性能问题：

• O(n) 对队列的操作 (queue.Contains(item) )) 对性能产生严重影响，因为队列增长
• 锁限制了并发性（他确实提到了）

以下代码通过

• 使用哈希集执行 O(1) 查找
• 改进了 Brian 的解决方案结合 2 个数据结构的 System.Collections.Concurrent 命名

空间由于没有 ConcurrentHashSet，我使用 ConcurrentDictionary，忽略这些值。

在这种罕见的情况下，幸运的是，可以简单地从多个更简单的数据结构中组合出更复杂的并发数据结构，而无需添加锁。两个并发数据结构上的操作顺序在这里很重要。

public class NoDuplicatesConcurrentQueue<T> : IProducerConsumerCollection<T>
{
    private readonly ConcurrentDictionary<T, bool> existingElements = new ConcurrentDictionary<T, bool>();
    private readonly ConcurrentQueue<T> queue = new ConcurrentQueue<T>();

    public bool TryAdd(T item)
    {
        if (existingElements.TryAdd(item, false))
        {
            queue.Enqueue(item);
            return true;
        }
        return false;
    }

    public bool TryTake(out T item)
    {
        if (queue.TryDequeue(out item))
        {
            bool _;
            existingElements.TryRemove(item, out _);
            return true;
        }
        return false;
    }
    ...
}

注意看待这个问题的另一种方式：您需要一个保留插入顺序的集合。

Answer 3

我建议使用锁来实现您的操作，这样您就不会以破坏项目的方式读写项目，从而使它们成为原子的。例如，对于任何 IEnumerable：

object bcLocker = new object();

// ...

lock (bcLocker)
{
    bool foundTheItem = false;
    foreach (someClass nextItem in myBlockingColl)
    {
        if (nextItem.Equals(item))
        {
            foundTheItem = true;
            break;
        }
    }
    if (foundTheItem == false)
    {
        // Add here
    }
}

Answer 4

如何访问阻塞集合的底层默认并发队列？

默认情况下，BlockingCollection 由 ConcurrentQueue 支持。换句话说，如果您没有显式指定其后备存储，它将在幕后创建一个 ConcurrentQueue。由于您希望直接访问底层存储，因此可以手动创建一个 ConcurrentQueue 并将其传递给 BlockingCollection 的构造函数：

ConcurrentQueue<Item> queue = new();
BlockingCollection<Item> collection = new(queue);

不幸的是ConcurrentQueue 集合没有 TryPeek 带有输入参数的方法，因此您想要做的事情是不可能的：

if (!queue.TryPeek(item)) // Compile error, missing out keyword
    collection.Add(item);

另请注意，队列现在由集合拥有。如果您尝试直接更改它（通过发出 Enqueue 或 TryDequeue 命令），collection 将引发异常。