ConcurrentQueue的Count属性的复杂度分析

之前程序中有如下代码：

private ConcurrentQueue<Tuple<DateTime, object>> recordq = new ConcurrentQueue<Tuple<DateTime, object>>();
private void Add2LogDetail(string symbol, object data)
{
    if (config.LogDic.Contains(symbol))
    {
        recordq.Enqueue(new Tuple<DateTime, object>(DateTime.Now, data));
        if (recordq.Count > 10000)
        {
            recordq.TryDequeue(out Tuple<DateTime, object> tmp);
            Logger.LogError("记录队列满，丢弃一笔:");
        }
    }
}

这段代码的主要记录指定股票的行情数据，将接收到的行情暂存在一个ConcurrentQueue里面，然后另外一个线程会从该队列里面获取数据然后持久化到本地磁盘。为了防止写线程出现异常而导致缓存队列被写爆，从而导致OOM，这里限制了recordq缓存的最大数据量为10000。这就涉及到需要频繁调用ConcurrentQueue的Count方法。然而这个Count方法的算法复杂度可能需要仔细研究一下。

在普通的在普通的非线程安全集合（如 List<T> 或 Queue<T>）中，Count 通常只是简单地返回一个内部维护的 int 字段，时间复杂度是严格的 O(1)。然而，当在并发集合中，为了追求极致的高吞吐量，设计的取舍可能会发生变化。本文通过分析 ConcurrentQueue<T> 的源码，分析ConcurrentQueue的Count实现，并尝试回答为什么它不使用简单的 Interlocked 计数。

ConcurrentQueue的内部结构

要理解Count，首先要理解ConcurrentQueue的数据结构，ConcurrentQueue不是一个简单的单链表，他是一个由环形数组组成了链表，它的内部是由多个Segment组成，每个Segment是一个固定大小的数组。

当队列不断Enqueue时，如果当前的Segment满了，就会分配一个新的Segment并链接到队尾。‘
当不断Dequeue时，如果一个Segment空了，就会被垃圾回收。

这种设计是为了减少垃圾回收的压力，并利用数组的CPU缓存友好型。

Count的源码分析

在 ConcurrentQueue<T> 的.NET Framework版本的源码中（.NET Core中的实现有些不一样），Count 属性的实现如下：

/// <summary>
/// Gets the number of elements contained in the <see cref="ConcurrentQueue{T}"/>.
/// </summary>
/// <value>The number of elements contained in the <see cref="ConcurrentQueue{T}"/>.</value>
/// <remarks>
/// For determining whether the collection contains any items, use of the <see cref="IsEmpty"/>
/// property is recommended rather than retrieving the number of items from the <see cref="Count"/>
/// property and comparing it to 0.
/// </remarks>
public int Count
{
    get
    {
        //store head and tail positions in buffer, 
        Segment head, tail;
        int headLow, tailHigh;
        GetHeadTailPositions(out head, out tail, out headLow, out tailHigh);

        if (head == tail)
        {
            return tailHigh - headLow + 1;
        }

        //head segment
        int count = SEGMENT_SIZE - headLow;

        //middle segment(s), if any, are full.
        //We don't deal with overflow to be consistent with the behavior of generic types in CLR.
        count += SEGMENT_SIZE * ((int)(tail.m_index - head.m_index - 1));

        //tail segment
        count += tailHigh + 1;

        return count;
    }
}

这里面重点在GetHeadTailPositions，获取一个一致的快照，得到相关的头部和尾部segment以及headLow和tailHigh。

/// <summary>
/// Store the position of the current head and tail positions.
/// </summary>
/// <param name="head">return the head segment</param>
/// <param name="tail">return the tail segment</param>
/// <param name="headLow">return the head offset, value range [0, SEGMENT_SIZE]</param>
/// <param name="tailHigh">return the tail offset, value range [-1, SEGMENT_SIZE-1]</param>
private void GetHeadTailPositions(out Segment head, out Segment tail,
    out int headLow, out int tailHigh)
{
    head = m_head;
    tail = m_tail;
    headLow = head.Low;
    tailHigh = tail.High;
    SpinWait spin = new SpinWait();

    //we loop until the observed values are stable and sensible.  
    //This ensures that any update order by other methods can be tolerated.
    while (
        //if head and tail changed, retry
        head != m_head || tail != m_tail
        //if low and high pointers, retry
        || headLow != head.Low || tailHigh != tail.High
        //if head jumps ahead of tail because of concurrent grow and dequeue, retry
        || head.m_index > tail.m_index)
    {
        spin.SpinOnce();
        head = m_head;
        tail = m_tail;
        headLow = head.Low;
        tailHigh = tail.High;
    }
}

可以看到，这里面为了保证并发情况下的准确性，GetHeadTailPositions 方法内部会自旋重试，以确保在读取首尾位置时，队列没有发生剧烈的结构性变化。因为队列是一个动态延伸的链表（Segment 链）。如果没有锁，假如读取了 Head 的位置（比如索引 10），正准备去读 Tail 的位置时，极高并发下可能瞬间发生了 10000 次入队和 10000 次出队。这时再去读 Tail，拿到的可能是一个极其离谱的数值，两者一减，Count 甚至可能变成负数或超大错误值。为了在不加锁的情况下获取“一致”的首尾快照，.NET 底层使用了无锁编程中经典的“读取-验证”循环（乐观并发控制），即重新读取一次各个变量，看它们有没有发生变化。如果都没有发生变化，说明在读取尾部的这极短时间内，队列没有发生“追尾”或者严重的段移动，这个快照是相对一致的，可以接受。否则自旋然后重试。

在极高并发的入队/出队场景下，GetHeadTailPositions中的while循环的验证可能会持续失败，导致这个 while 循环不断自旋（消耗 CPU）。这就是为什么 Count 在高并发下不仅慢，还会浪费 CPU 资源的核心原因。

从源码可以看出，如果队列并发程度很低，那么Count的复杂度是 O(1) 的。但如果并发程度很高，Count内部调用的GetHeadTailPositions 方法会自旋重试，以确保在读取首尾位置时，队列没有发生剧烈的结构性变化，这意味着在极高并发下，调用 Count 会导致性能损耗。

为什么要这样实现？

有一个很简单的办法来实现Count，那就是在非并发集合中常见的那样，内部维护一个变量count，入队时 Interlocked.Increment，出队时 Interlocked.Decrement，Count 马上就能变成完美的 O(1)。ConcurrentQueue中的Count之所以没这么实现，我猜想可能有如下原因：

缓存失效：现代 CPU 都有 L1/L2 缓存。当一个线程修改了一个变量（比如那个全局的 int count），CPU 必须通过缓存一致性协议（如 MESI）广播一条消息，使得其他所有核心中缓存了该变量的“缓存行（Cache Line）”全部失效。
串行化：ConcurrentQueue 的核心卖点是无锁（Lock-free）和高并发。它巧妙地利用了分离的头尾指针，使得出队和入队操作几乎互不干扰。如果引入了一个全局的 count 变量，那么所有正在执行入队和出队的线程，都会在物理硬件层面为了争夺这同一个 count 变量所在的缓存行而发生冲突。这实质上把本该并行执行的操作，在硬件级别变成了串行操作。

ConcurrentQueue的设计目的优先考虑的是告诉收发数据，而不是用来计数的，所以为了保证Enqueue和Dequeue的吞吐，牺牲了Count的性能。

一种实现方法

如果实在要获得ConcurrentQueue的数量，可以在外部定义一个int，然后使用Interlocked来原子递增和递减来实现计数：

ConcurrentQueue<int> _queue = new();
int _fastCount = 0;

public void Enqueue(int item)
{
    _queue.Enqueue(item);
    Interlocked.Increment(ref _fastCount);
}

public bool TryDequeue(out int item)
{
    if (_queue.TryDequeue(out item))
    {
        Interlocked.Decrement(ref _fastCount);
        return true;
    }
    return false;
}

优缺点也很明显，优点在于读取 _fastCount 是严格的 O(1) 操作，且不涉及任何遍历。如果你有一个监控线程每秒要读取几万次 Count，这种写法能避免底层 Segment 遍历的开销。缺点就是如前所述，在入队和出队时，都要访问共同的_fastCount对象，另外 Count 和内部的队列状态无法保持强一致性，因为操作_queue和操作_fastCount不是一个原子操作，比如在执行_queue.Enqueue后，正准备执行Interlock.Increment时，系统被挂起，此时队列里有元素，但_fastCount仍然是0。另外，在代码所有的地方，只要有操作Enqueue和Dequeue，都需要同步进行_fastCount的变更。

因此只有在需要频繁访问Count的情况下，这种做法才更好。

总结

在使用集合时，如果要判断是否为空，使用IsEmpty或者Any()方法，在大多数集合中，是否为空是一个极其快速的O(1)操作，以ConcurrentQueue为例，他只需要检查头尾指针的状态。绝对不要写if (queue.Count>0)这样的代码。另外，不是所有的集合的Count方法的时间复杂度都是相同的，主要要考虑的是集合内部实现的数据结构。最后，在并发系统中，由于时差的存在，当拿到Count的那一刻，队列的真实数量肯呢个已经发生改变，所以Count只能用作日志记录、监控指标或负载均衡的参考值。

码农兼一个普普通通小青年

ConcurrentQueue的内部结构

Count的源码分析

为什么要这样实现？

一种实现方法

总结