并发集合和独特元素

Question

我有一个包含重复元素的并发 BlockingCollection 。如何修改它以添加或获取不同的元素？

Answer 1

BlockingCollection 的默认后备存储是 ConcurrentQueue。正如其他人指出的那样，使用它来添加不同的项目相当困难。

不过，您可以创建自己的实现 IProducerConsumerCollection 的集合类型，并将其传递给 BlockingCollection 构造函数。

想象一个 ConcurrentDictionary，其中包含当前队列中项目的键。要添加项目，请首先在字典上调用 TryAdd，如果该项目不在字典中，则将其添加，并将其添加到队列中。 Take（和 TryTake）从队列中获取下一个项目，将其从字典中删除，然后返回。

我希望有一个并发的 HashTable，但由于没有并发的，所以您必须使用 ConcurrentDictionary。

Answer 2

这是 IProducerConsumerCollection< 的实现;T> 具有队列行为的集合，也拒绝重复的项目：

public class ConcurrentQueueNoDuplicates<T> : IProducerConsumerCollection<T>
{
    private readonly Queue<T> _queue = new();
    private readonly HashSet<T> _set;
    private object Locker => _queue;

    public ConcurrentQueueNoDuplicates(IEqualityComparer<T> comparer = default)
    {
        _set = new(comparer);
    }

    public bool TryAdd(T item)
    {
        lock (Locker)
        {
            if (!_set.Add(item))
                throw new DuplicateKeyException();
            _queue.Enqueue(item); return true;
        }
    }

    public bool TryTake(out T item)
    {
        lock (Locker)
        {
            if (_queue.Count == 0)
                throw new InvalidOperationException();
            item = _queue.Dequeue();
            bool removed = _set.Remove(item);
            Debug.Assert(removed);
            return true;
        }
    }

    public int Count { get { lock (Locker) return _queue.Count; } }
    public bool IsSynchronized => false;
    public object SyncRoot => throw new NotSupportedException();
    public T[] ToArray() { lock (Locker) return _queue.ToArray(); }
    public IEnumerator<T> GetEnumerator() => ToArray().AsEnumerable().GetEnumerator();
    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
    public void CopyTo(T[] array, int index) => throw new NotSupportedException();
    public void CopyTo(Array array, int index) => throw new NotSupportedException();
}

public class DuplicateKeyException : InvalidOperationException { }

使用示例：

BlockingCollection<Item> queue = new(new ConcurrentQueueNoDuplicates<Item>());

//...

try { queue.Add(item); }
catch (DuplicateKeyException) { Console.WriteLine($"The {item} was rejected."); }

警告：调用如果项目重复，queue.TryAdd(item); 不会出现返回 false 的预期行为。任何添加重复项的尝试都必然会导致 DuplicateKeyException。不要尝试通过返回 false 来“修复”上述 ConcurrentQueueNoDuplicates.TryAdd 实现或 TryTake。 BlockingCollection将通过抛出不同的异常 (InvalidOperationException) 做出反应，并且最重要的是其内部状态将被损坏。目前（.NET 7）一个错误，该错误会将 BlockingCollection 的有效容量减少 1，其底层存储具有 TryAdd 实现，该实现返回false。该错误已修复< /a> 对于 .NET 8，这将防止损坏，但不会改变错误抛出行为。