C#并发集合-ConcurrentQueue

在多线程环境下，使用这个集合接受网络请求

using System.Collections; using System.Collections.Concurrent; using System.Runtime.Serialization; namespace ConsoleApp1.Concurrent; /* * 并发队列 ConcurrentQueue * 数据结构：使用 数组 + 单链表 * 结构图： * tail head * segment1 -> segment2 -> ... -> segment... * arr[0] arr[0] arr[0] * arr[1] arr[1] arr[1] * arr[...] arr[...] arr[...] * arr[32] arr[32] arr[32] * 操作： * 进队 * tail => * 累加 tail segment 中的 high（入队方向从 0 到 31） * 入队完成，if segment 内数组已满（high == 31），则提前将下一个 segment 创建，并且当前 tail.next 指向新的segment, tail 再执行 新的segment * 出队 * head => * 累加 head segment 中的 low（出队方向从 0 到 31） * if segment 内数组已空（low == 31），则提前将下一个 segment 创建，并且当前 tail.next 指向新的segment, tail 再执行 新的segment */ public class ConcurrentQueue<T> : IProducerConsumerCollection<T> { private volatile ConcurrentQueue<T>.Segment m_head; private volatile ConcurrentQueue<T>.Segment m_tail; internal volatile int m_numSnapshotTakers; private const int SEGMENT_SIZE_32 = 32; public ConcurrentQueue() { this.m_head = this.m_tail = new Segment(0L, this); } private void InitializeFromCollection(IEnumerable<T> collection) { throw new NotImplementedException(); } public ConcurrentQueue(IEnumerable<T> collection) { throw new NotImplementedException(); } private void OnSerializing(StreamingContext context) { throw new NotImplementedException(); } private void OnSerialized(StreamingContext context) { throw new NotImplementedException(); } void ICollection.CopyTo(Array array, int index) { throw new NotImplementedException(); } private IEnumerator<T> GetEnumerator(Segment head, Segment tail, int headLow, int tailHigh) { throw new NotImplementedException(); } public IEnumerator<T> GetEnumerator() { throw new NotImplementedException(); } IEnumerator IEnumerable.GetEnumerator() => GetEnumerator(); public void CopyTo(Array array, int index) { throw new NotImplementedException(); } public int Count { get { this.GetHeadTailPositions(out Segment head, out Segment tail, out int headLow, out int tailHigh); return head == tail ? tailHigh - headLow + 1 : SEGMENT_SIZE_32 - headLow + SEGMENT_SIZE_32 * (int)(tail.m_index - head.m_index - 1L) + (tailHigh + 1); } } private void GetHeadTailPositions(out Segment head, out Segment tail, out int headLow, out int tailHigh) { head = this.m_head; tail = this.m_tail; headLow = head.Low; tailHigh = tail.High; SpinWait spinWait = new SpinWait(); while (head != this.m_head || tail != this.m_tail || headLow != head.Low || tailHigh != tail.High || head.m_index > tail.m_index) { spinWait.SpinOnce(); head = this.m_head; tail = this.m_tail; headLow = head.Low; tailHigh = tail.High; } } public bool IsSynchronized => false; public object SyncRoot => throw new NotSupportedException(); public void CopyTo(T[] array, int index) { throw new NotImplementedException(); } public bool TryAdd(T item) { this.Enqueue(item); return true; } public bool TryTake(out T item) => this.TryDequeue(out item); public void Enqueue(T item) { SpinWait spinWait = new SpinWait(); while (!this.m_tail.TryAppend(item)) spinWait.SpinOnce(); } public bool TryDequeue(out T result) { while (!this.IsEmpty) { if (this.m_head.TryRemove(out result)) return true; } result = default(T); return false; } public bool TryPeek(out T result) { Interlocked.Increment(ref this.m_numSnapshotTakers); while (!this.IsEmpty) { if (this.m_head.TryPeek(out result)) { Interlocked.Decrement(ref this.m_numSnapshotTakers); return true; } } result = default(T); Interlocked.Decrement(ref this.m_numSnapshotTakers); return false; } public bool IsEmpty { get { ConcurrentQueue<T>.Segment head = this.m_head; if (!head.IsEmpty) return false; if (head.Next == null) return true; SpinWait spinWait = new SpinWait(); for (; head.IsEmpty; head = this.m_head) { if (head.Next == null) return true; spinWait.SpinOnce(); } return false; } } public T[] ToArray() { throw new NotImplementedException(); } internal struct VolatileBool { public volatile bool m_value; public VolatileBool(bool value) => this.m_value = value; } private class Segment { // 存在实际的元素 internal volatile T[] m_array; internal volatile VolatileBool[] m_state; // 单链表，指向下一个 segment internal ConcurrentQueue<T>.Segment Next => this.m_next; private volatile ConcurrentQueue<T>.Segment m_next; // 记录当前 segment 为 第几个（从0开始，计算总Count 会使用） internal readonly long m_index; // 当前 segment 中 的 可出队下标 internal int Low => Math.Min(this.m_low, SEGMENT_SIZE_32); private volatile int m_low; // 当前 segment 中 的 可进队下标 internal int High => Math.Max(this.m_high, 31); private volatile int m_high; // 属于哪个队列 private volatile ConcurrentQueue<T> m_source; internal Segment(long index, ConcurrentQueue<T> source) { this.m_array = new T[SEGMENT_SIZE_32]; this.m_state = new VolatileBool[SEGMENT_SIZE_32]; this.m_high = -1; this.m_index = index; this.m_source = source; } internal bool IsEmpty => this.Low > this.High; // 创建下一个 segment internal void Grow() { this.m_next = new Segment(this.m_index + 1L, this.m_source); // 将新创建的 segment 设置为 当前 队尾 this.m_source.m_tail = this.m_next; } internal bool TryAppend(T value) { // 当前 segment 队尾 已经到头，无法添加元素 if (this.m_high >= 31) return false; int high = SEGMENT_SIZE_32; try { } finally { // 原值性的增加队尾下标 high = Interlocked.Increment(ref this.m_high); if (high <= 31) { // 设置队尾元素 this.m_array[high] = value; // 设置元素添加标志（由于 value 可能是任意值，这里使用额外的bool 表示是否添加） this.m_state[high].m_value = true; } // 队尾 已经 到头，开始创建一下 segment if (high == 31) this.Grow(); } // 如果抢到的下标超过，则直接返回失败 return high <= 31; } internal bool TryRemove(out T result) { SpinWait spinWait1 = new SpinWait(); int low = this.Low; for (int high = this.High; low <= high; high = this.High) { // 原子性的尝试对 这个 segment 的 队头下标 + 1，这里可能多个线程同时抢占 if (Interlocked.CompareExchange(ref this.m_low, low + 1, low) == low) { SpinWait spinWait2 = new SpinWait(); // 当前存在进队元素，但是尚未完成进队 的全部操作，进队操作完成 m_value == true while (!this.m_state[low].m_value) spinWait2.SpinOnce(); // 获取这个队列元素 result = this.m_array[low]; /* * 是否有操作持有快照，没有则直接赋值元素为默认值 */ if (this.m_source.m_numSnapshotTakers <= 0) this.m_array[low] = default(T); /* * 如果已经取到 当前 segment 队列的 队头 => low 到达 数组的最后一个下标 * Note:这里为什么不能写成 if(low == 31)?, 其实是可以的 */ if (low + 1 >= SEGMENT_SIZE_32) { // 等待进入下一个 segment，根据 入队函数可知，当一个segment 队头达到最后一个位置后，会立马创建 下一个 segment SpinWait spinWait3 = new SpinWait(); while (this.m_next == null) spinWait3.SpinOnce(); // 将下一个 segment 设置为 队头 segment this.m_source.m_head = this.m_next; } // 抢到 队列中的数据 return true; } spinWait1.SpinOnce(); // 前面 没有 抢到 队列中的数据，重新获取当前的最新 Low => 队头下标 low = this.Low; } // 未抢到 队列中的数据，直接返回失败 result = default(T); return false; } internal bool TryPeek(out T result) { result = default(T); int low = this.Low; // 队列无元素 if (low > this.High) return false; // 当前队头元素，入队尚未完成，自旋一会 SpinWait spinWait = new SpinWait(); while (!this.m_state[low].m_value) spinWait.SpinOnce(); // 获取当前队头元素 result = this.m_array[low]; return true; } } }