using System; using System.Collections; using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Runtime.InteropServices; using System.Threading; namespace Masuit.Tools.Systems; /// /// 线程安全的唯一元素队列,结合了ConcurrentQueue和ConcurrentDictionary的特性 /// /// 队列中元素的类型 [DebuggerDisplay("Count = {Count}")] [ComVisible(false)] public class ConcurrentHashQueue : IProducerConsumerCollection, IReadOnlyCollection, ICollection, ICollection { private class Node { public readonly T Value; public Node Next; public Node(T value) { Value = value; } } private volatile Node _head; private volatile Node _tail; private readonly ConcurrentHashSet _hashSet; private readonly ReaderWriterLockSlim _lock = new(LockRecursionPolicy.SupportsRecursion); private int _count; #region 构造函数 /// /// 初始化 ConcurrentUniqueQueue 类的新实例 /// public ConcurrentHashQueue() : this(null, null) { } /// /// 初始化 ConcurrentUniqueQueue 类的新实例,使用指定的相等比较器 /// /// 用于比较值的 IEqualityComparer public ConcurrentHashQueue(IEqualityComparer comparer) : this(null, comparer) { } /// /// 初始化 ConcurrentUniqueQueue 类的新实例,包含从指定集合复制的元素 /// /// 要复制到 ConcurrentUniqueQueue 中的集合 public ConcurrentHashQueue(IEnumerable collection) : this(collection, null) { } /// /// 初始化 ConcurrentUniqueQueue 类的新实例,包含从指定集合复制的元素,并使用指定的相等比较器 /// /// 要复制到 ConcurrentUniqueQueue 中的集合 /// 用于比较值的 IEqualityComparer public ConcurrentHashQueue(IEnumerable collection, IEqualityComparer comparer) { _hashSet = new ConcurrentHashSet(comparer ?? EqualityComparer.Default); // 创建哨兵节点 _head = _tail = new Node(default(T)); _count = 0; if (collection != null) { foreach (var item in collection) { Enqueue(item); } } } #endregion 构造函数 #region 核心方法 /// /// 将对象添加到 ConcurrentUniqueQueue 的结尾处 /// /// 要添加到 ConcurrentUniqueQueue 的对象 /// 如果元素成功添加返回 true,如果元素已存在则返回 false public bool Enqueue(T item) { if (!_hashSet.TryAdd(item)) return false; // 创建新节点 var newNode = new Node(item); // 使用原子操作添加节点到链表尾部 while (true) { var tail = _tail; var next = tail.Next; if (tail == _tail) { if (next == null) { // 尝试链接新节点 if (Interlocked.CompareExchange(ref tail.Next, newNode, null) == null) { // 尝试移动tail指针 Interlocked.CompareExchange(ref _tail, newNode, tail); Interlocked.Increment(ref _count); return true; } } else { // 帮助其他线程移动tail指针 Interlocked.CompareExchange(ref _tail, next, tail); } } } } /// /// 尝试移除并返回位于 ConcurrentUniqueQueue 开始处的对象 /// /// 如果操作成功,则返回已移除的对象;否则为 T 的默认值 /// 如果成功移除并返回了对象,则为 true;否则为 false public bool TryDequeue(out T result) { while (true) { var head = _head; var tail = _tail; var next = head.Next; if (head == _head) { if (head == tail) { if (next == null) { result = default(T); return false; } // 帮助其他线程移动tail指针 Interlocked.CompareExchange(ref _tail, next, tail); } else { result = next.Value; // 尝试移动head指针 if (Interlocked.CompareExchange(ref _head, next, head) == head) { // 从set中移除元素 _hashSet.Remove(result); Interlocked.Decrement(ref _count); return true; } } } } } /// /// 尝试返回位于 ConcurrentUniqueQueue 开始处的对象但不将其移除 /// /// 如果操作成功,则返回开头的对象;否则为 T 的默认值 /// 如果成功返回了对象,则为 true;否则为 false public bool TryPeek(out T result) { while (true) { var head = _head; var next = head.Next; if (next == null) { result = default(T); return false; } if (head == _head) { result = next.Value; return true; } } } /// /// 确定某元素是否在 ConcurrentUniqueQueue 中 /// /// 要在 ConcurrentUniqueQueue 中定位的对象 /// 如果在 ConcurrentUniqueQueue 中找到 item,则为 true;否则为 false public bool Contains(T item) { return _hashSet.Contains(item); } /// Adds an item to the . /// The object to add to the . /// The is read-only. public void Add(T item) { Enqueue(item); } /// /// 从 ConcurrentUniqueQueue 中移除所有对象 /// public void Clear() { _lock.EnterWriteLock(); try { _hashSet.Clear(); _head = _tail = new Node(default(T)); _count = 0; } finally { if (_lock.IsWriteLockHeld) { _lock.ExitWriteLock(); } } } #endregion 核心方法 #region 属性和状态 /// Removes the first occurrence of a specific object from the . /// The object to remove from the . /// true if item was successfully removed from the ; otherwise, false. This method also returns false if item is not found in the original . /// The is read-only. public bool Remove(T item) { _lock.EnterWriteLock(); try { // 双重检查,确保元素仍然存在 if (!_hashSet.Contains(item)) return false; // 从HashSet中移除 _hashSet.Remove(item); // 遍历链表重新构建,跳过要移除的元素 var newHead = new Node(default(T)); var newTail = newHead; var current = _head.Next; var found = false; while (current != null) { if (!_hashSet.Comparer.Equals(current.Value, item)) { newTail.Next = new Node(current.Value); newTail = newTail.Next; } else { found = true; } current = current.Next; } if (found) { _head = newHead; _tail = newTail; Interlocked.Decrement(ref _count); return true; } // 如果没有找到,重新添加到HashSet(理论上不应该发生) _hashSet.TryAdd(item); return false; } finally { if (_lock.IsWriteLockHeld) { _lock.ExitWriteLock(); } } } /// /// 获取 ConcurrentUniqueQueue 中包含的元素数 /// public int Count => _count; /// Gets a value indicating whether the is read-only. /// true if the is read-only; otherwise, false. public bool IsReadOnly { get; } /// /// 获取一个值,该值指示 ConcurrentUniqueQueue 是否为空 /// public bool IsEmpty => _head.Next == null; #endregion 属性和状态 #region IProducerConsumerCollection 实现 /// /// 将项添加到 IProducerConsumerCollection /// /// 要添加到 IProducerConsumerCollection 的项 /// 如果成功添加了项,则为 true;否则为 false bool IProducerConsumerCollection.TryAdd(T item) { return Enqueue(item); } /// /// 尝试从 IProducerConsumerCollection 中移除和返回对象 /// /// 此方法返回时,如果成功移除并返回了对象,则 item 包含所移除的对象 /// 如果成功移除并返回了对象,则为 true;否则为 false bool IProducerConsumerCollection.TryTake(out T item) { return TryDequeue(out item); } /// /// 将 IProducerConsumerCollection 的元素复制到新数组 /// /// 包含从 IProducerConsumerCollection 复制的元素的新数组 public T[] ToArray() { var snapshot = new List(Count); var current = _head.Next; while (current != null) { snapshot.Add(current.Value); current = current.Next; } return snapshot.ToArray(); } /// /// 将 IProducerConsumerCollection 的元素复制到现有一维数组中 /// /// 一维数组,它是从 IProducerConsumerCollection 复制的元素的目标 /// array 中从零开始的索引,从此处开始复制 public void CopyTo(T[] array, int index) { if (array == null) throw new ArgumentNullException(nameof(array)); if (index < 0) throw new ArgumentOutOfRangeException(nameof(index), "索引不能为负数。"); if (index >= array.Length) throw new ArgumentOutOfRangeException(nameof(index), "索引等于或大于数组长度。"); var snapshot = ToArray(); if (snapshot.Length > array.Length - index) throw new ArgumentException("目标数组空间不足。"); Array.Copy(snapshot, 0, array, index, snapshot.Length); } /// /// 返回循环访问 ConcurrentUniqueQueue 的枚举数 /// /// ConcurrentUniqueQueue 的枚举数 public IEnumerator GetEnumerator() { var current = _head.Next; while (current != null) { yield return current.Value; current = current.Next; } } IEnumerator IEnumerable.GetEnumerator() { return GetEnumerator(); } #endregion IProducerConsumerCollection 实现 #region ICollection 实现 /// /// 从特定的 ICollection 索引处开始,将数组的元素复制到一个数组中 /// /// 一维数组,它是从 ICollection 复制的元素的目标 /// array 中从零开始的索引,从此处开始复制 void ICollection.CopyTo(Array array, int index) { if (array == null) throw new ArgumentNullException(nameof(array)); if (array.Rank != 1) throw new ArgumentException("仅支持一维数组。"); if (index < 0 || index >= array.Length) throw new ArgumentOutOfRangeException(nameof(index)); var tArray = array as T[]; if (tArray != null) { CopyTo(tArray, index); return; } var elementType = array.GetType().GetElementType(); if (!elementType.IsAssignableFrom(typeof(T))) throw new ArgumentException("数组类型与集合元素类型不兼容。"); var objects = array as object[]; if (objects == null) throw new ArgumentException("目标数组类型不受支持。"); try { var snapshot = ToArray(); Array.Copy(snapshot, 0, objects, index, snapshot.Length); } catch (ArrayTypeMismatchException) { throw new ArgumentException("数组类型与集合元素类型不兼容。"); } } /// /// 获取一个值,该值指示是否同步对 ICollection 的访问 /// bool ICollection.IsSynchronized => false; /// /// 获取可用于同步对 ICollection 的访问的对象 /// object ICollection.SyncRoot => _lock; #endregion ICollection 实现 #region 批量操作方法 /// /// 批量入队多个元素 /// /// 要入队的元素集合 /// 成功添加的元素数量 public int EnqueueRange(IEnumerable items) { if (items == null) throw new ArgumentNullException(nameof(items)); var count = 0; foreach (var item in items) { if (Enqueue(item)) { count++; } } return count; } /// /// 批量出队多个元素 /// /// 要出队的元素数量 /// 出队的元素集合 public IEnumerable DequeueRange(int count) { if (count < 0) throw new ArgumentOutOfRangeException(nameof(count), "数量不能为负数。"); var result = new List(Math.Min(count, Count)); for (int i = 0; i < count && TryDequeue(out T item); i++) { result.Add(item); } return result; } #endregion 批量操作方法 #region 原子操作方法 /// /// 原子性地入队元素,如果队列已满则返回false /// /// 要入队的元素 /// 队列的最大大小 /// 如果成功入队返回true,否则返回false public bool TryEnqueueWithLimit(T item, int maxSize) { if (maxSize <= 0) throw new ArgumentOutOfRangeException(nameof(maxSize), "最大大小必须大于0。"); // 先检查是否已达到最大大小 if (Count >= maxSize && !Contains(item)) return false; return Enqueue(item); } /// /// 原子性地入队元素,如果队列已满则移除最旧的元素 /// /// 要入队的元素 /// 队列的最大大小 /// 被移除的元素,如果没有移除则返回default(T) public T EnqueueWithEviction(T item, int maxSize) { if (maxSize <= 0) throw new ArgumentOutOfRangeException(nameof(maxSize), "最大大小必须大于0。"); T evicted = default(T); // 如果元素已存在,先移除旧的实例 if (Contains(item)) { RemoveExistingAndEnqueue(item); return default(T); } // 检查是否需要驱逐 if (Count >= maxSize) { TryDequeue(out evicted); } Enqueue(item); return evicted; } /// /// 移除已存在的元素并重新入队(用于更新元素位置) /// /// 要更新的元素 /// 如果成功更新返回true,否则返回false private bool RemoveExistingAndEnqueue(T item) { // 这个操作比较复杂,需要锁定整个队列 _lock.EnterWriteLock(); try { if (!_hashSet.Contains(item)) return false; // 从字典中暂时移除 _hashSet.Remove(item); // 重新构建链表,跳过要更新的元素 var newHead = new Node(default(T)); var newTail = newHead; var current = _head.Next; var found = false; while (current != null) { if (!_hashSet.Comparer.Equals(current.Value, item)) { newTail.Next = new Node(current.Value); newTail = newTail.Next; _hashSet.TryAdd(current.Value); } else { found = true; } current = current.Next; } if (found) { // 添加新元素到尾部 newTail.Next = new Node(item); newTail = newTail.Next; _hashSet.TryAdd(item); _head = newHead; _tail = newTail; _count = _hashSet.Count; return true; } return false; } finally { if (_lock.IsWriteLockHeld) { _lock.ExitWriteLock(); } } } #endregion 原子操作方法 }