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 原子操作方法
}