ConcurrentNativeQueue<T>：一个使用 .NET 实现的零 GC 压力的无锁 MPSC 原生队列

一、为什么要造这个轮子

.NET 提供了 ConcurrentQueue<T> 和 Channel<T> 两种开箱即用的并发队列。对大多数业务场景，它们已经足够好。但在以下场景中，它们的底层设计决策会成为性能瓶颈：

• 游戏主循环 / 音频管线：GC 停顿（即使是 Gen0）会导致可感知的帧卡顿或音频爆音。即使 Workstation GC 的 Gen0 暂停只有 ~100μs，在 16ms 的帧预算中也可能造成掉帧。
• 高频交易 / 实时数据管线：每微秒都有价值，托管堆分配意味着不可预测的 GC 介入。
• Native interop 密集场景：数据需要频繁在 managed/unmanaged 边界传递，如果队列本身就在 native 内存上，可以省去 pin/copy 开销。
• AOT 发布：ConcurrentNativeQueue<T> 是纯 unmanaged 结构体，天然适合 NativeAOT 场景。

ConcurrentNativeQueue<T> 的目标很明确：在 MPSC（多生产者单消费者）模式下，提供零 GC 压力、零托管堆分配的高吞吐量队列。这不是要替代 ConcurrentQueue<T>，而是为那些"对 GC 停顿零容忍"的场景提供一个专用工具。

二、整体架构

ConcurrentNativeQueue<T> (struct, unmanaged)
├── _head ──→ [SegmentHeader*] ──→ [SegmentHeader*] ──→ ...
│                  ↓                     ↓
│              [Slot* 数组]          [Slot* 数组]
│              (已消费,释放)         (消费中)
│
├── _tail ──→ [SegmentHeader*] ──→ (预建的下一段)
│                  ↓
│              [Slot* 数组]
│              (生产中)
│
├── _origin ──→ 首段（用于 Dispose 遍历释放所有段头）
│
└── 缓存行填充：_head/_dequeuePos 与 _tail 之间 64 字节隔离

所有内存（段头结构体 + 槽位数组）均通过 NativeMemory 分配。ConcurrentNativeQueue<T> 本身也是 struct，整个生命周期不产生任何托管堆分配。

三、核心技术原理

3.1 无锁入队：Volatile.Read + CAS

入队操作不使用锁，核心路径只有一次原子操作：

// 1. 纯读检测：当前段是否有空位
long pos = Volatile.Read(ref tail->EnqueuePos.Value);
long offset = pos - tail->BaseIndex;
if (offset >= tail->Capacity) { /* 段满，推进到下一段 */ }

// 2. CAS 占位
if (Interlocked.CompareExchange(ref tail->EnqueuePos.Value, pos + 1, pos) == pos)
{
    // 3. 写入数据，设置标记
    tail->Slots[offset].Value = item;
    Volatile.Write(ref tail->Slots[offset].State, 1);
}

关键设计点：

• 段满检测是纯读操作（Volatile.Read），不产生任何原子写。多个生产者同时检测到段满时，只有 read 竞争，不会弄脏 cache line。
• CAS 只在段有空位时才执行。段满时 Volatile.Read 发现 offset >= Capacity 后直接走段切换路径，避免了无效的原子写。
• 每次入队只有 1 次原子操作（CAS）。

3.2 无锁出队：单消费者的极致简化

因为约定了单消费者（MPSC 的 SC 部分），出队路径不需要任何原子操作：

// 纯本地读：检查 State 标记
if (Volatile.Read(ref head->Slots[offset].State) != 1)
    return false; // 数据还没准备好

item = head->Slots[offset].Value;
_dequeuePos = pos + 1; // 普通写，无需原子操作

_dequeuePos 是消费者的私有字段，只有单消费者读写，所以普通赋值即可。这比 ConcurrentQueue<T> 的 MPMC 出队路径（需要 CAS）快得多。

3.3 分段链表 + 指数增长

队列不使用单一连续缓冲区，而是由多个段组成的链表。每个段是一个固定大小的原生内存数组：

[段 0: 32 slots] → [段 1: 64 slots] → [段 2: 128 slots] → ... → [段 N: 1M slots]
                                          容量指数增长，上限 1M

段满时，生产者通过 CAS 创建新段并链接：

// 只有一个生产者能成功链接新段
if (Interlocked.CompareExchange(ref seg->Next, (nint)newSeg, (nint)0) != (nint)0)
{
    // CAS 失败：另一个生产者先创建了，释放多余段
    NativeMemory.Free(newSeg->Slots);
    NativeMemory.Free(newSeg);
}

指数增长的收益：处理 1 亿条数据，段切换只发生约 20 次（32+64+128+...+1M+1M+...）。而固定 32 大小需要 300 多万次段切换。

3.4 预建下一段

当段接近满时（剩余 ≤ 16 个槽位），生产者在写入成功后提前分配下一段：

if (offset + 1 >= tail->Capacity - PreBuildSlots
    && Volatile.Read(ref tail->Next) == (nint)0)
    EnsureNextSegment(tail);

这样当段真正满时，新段已经准备好了，TryAdvanceTail 只需一次 CAS 推进尾指针。代价是每次 Enqueue 多一个必定被预测为 false 的分支（~0 额外开销）。

3.5 两阶段内存回收

这是全 native 化设计中最核心的生命周期问题。

为什么段头不能在消费后立即释放？ 因为生产者可能在任意时刻被 OS 抢占，此时它持有旧段的裸指针。如果消费者释放了这个段头，生产者恢复后会访问已释放的内存（use-after-free）。

解决方案：

阶段	释放内容	时机	安全性依据
阶段 1	槽位数组（Slot*）	消费者消费完整段后立即释放	所有 State == 1 → 所有生产者已写完并离开该槽位
阶段 2	段头结构体（SegmentHeader*）	Dispose() 时从 _origin 遍历释放	队列不再使用，无并发访问

段头只有 ~160 字节，指数增长使段头数量为对数级别：处理 10 亿条数据 ≈ 1000 个段头 ≈ 200KB，完全可忽略。

3.6 False Sharing 防护

在多核 CPU 上，如果生产者和消费者的热点字段位于同一条 64 字节缓存行上，每次写入都会导致对方核心的缓存行失效（false sharing），严重降低性能。

// ── 消费者缓存行 ──
private SegmentHeader* _head;       // 消费者读写
private long _dequeuePos;           // 消费者读写
private SegmentHeader* _origin;

private long _p0, _p1, _p2, _p3, _p4, _p5, _p6, _p7;  // 64 字节填充

// ── 生产者缓存行 ──
private nint _tail;                 // 生产者读写

同时，段内的 EnqueuePos（生产者热点）使用 PaddedLong（128 字节，值在偏移 64 处），确保不与段头的只读元数据（Capacity、BaseIndex 等）共享缓存行。

四、与 ConcurrentQueue<T> 的对比

维度	ConcurrentQueue<T>	ConcurrentNativeQueue<T>
并发模型	MPMC（多生产者多消费者）	MPSC（多生产者单消费者）
内存分配	托管 T[]，由 GC 管理	NativeMemory，手动管理
GC 压力	段数组进入托管堆，GC 需标记	零：不产生任何托管堆分配
出队路径	需要 CAS（支持多消费者）	纯本地读写（单消费者无竞争）
入队原子操作	1 次 CAS（Interlocked.Inc）	1 次 CAS（CompareExchange）
类型约束	任意 T	where T : unmanaged
生命周期	自动（GC 回收）	需调用 Dispose()
适用场景	通用并发队列	GC 敏感 / 实时 / Native interop
段大小	指数增长（32 → 1M）	指数增长（32 → 1M）

核心权衡：ConcurrentNativeQueue<T> 用 MPSC 约束 + unmanaged 约束 + 手动 Dispose，换取了零 GC 压力和更快的出队路径。如果你的场景不需要对抗 GC 停顿，ConcurrentQueue<T> 是更好的选择——经过微软多年打磨，支持 MPMC，不需要 unsafe。

五、应用场景

适合使用的场景

• 游戏引擎消息总线：多个系统（物理、AI、网络）作为生产者，主线程渲染循环作为单消费者。帧循环中零 GC 停顿。
• 音频处理管线：音频回调线程作为消费者，其他线程推送音频事件。音频线程对延迟极其敏感。
• 日志收集器：多个业务线程写日志，单个后台线程批量刷盘。高吞吐量 + 零 GC 是理想特性。
• Native interop 数据桥：与 C/C++ 库交互时，数据直接在 native 内存上传递，免去 pin/copy。
• NativeAOT 应用：ConcurrentNativeQueue<T> 是纯 unmanaged struct，完美兼容 AOT 发布。

不适合的场景

• 需要多消费者（MPMC）
• T 包含引用类型（string、class）
• 不愿意管理 Dispose 生命周期
• 对 GC 停顿不敏感的一般业务逻辑

六、优缺点总结

优点

1. 零 GC 压力：所有内存通过 NativeMemory 分配，GC 完全不感知
2. 低延迟出队：单消费者无需原子操作，纯本地读写
3. 高吞吐量入队：CAS 无锁 + 段满纯读检测 + 预建下一段
4. 批量入队：EnqueueRange 单次 CAS 占位 N 个槽位
5. 自适应段大小：指数增长减少段切换，自动适应负载
6. False sharing 防护：缓存行填充隔离生产者/消费者热点
7. unmanaged struct：结构体本身可嵌入其他 native 数据结构

缺点

1. 仅支持单消费者（MPSC 约束）
2. 仅支持 unmanaged 类型（不能存 string、object）
3. 必须手动 Dispose（否则内存泄漏）
4. 需要 unsafe（项目必须启用不安全代码）
5. 段头延迟释放（Dispose 前累积，虽然开销极小）

七、基准测试

测试环境

• CPU: Intel Core i7-14700K (20C/28T)
• Runtime: .NET 10.0.3, X64 RyuJIT x86-64-v3
• OS: Windows 10
• 数据量: 1,000,000 条/轮（long 类型）
• 同步方式: 专用 Thread + ManualResetEventSlim 三阶段同步

MPSC 吞吐量对比（MpscBenchmark）

Method	ProducerCount	Mean	Error	StdDev	Ratio	Allocated	Alloc Ratio
ConcurrentQueue	1	10.87 ms	0.295 ms	0.857 ms	1.01	1,050,600 B	1.000
ConcurrentNativeQueue	1	18.14 ms	0.340 ms	0.334 ms	1.68	488 B	0.000
UnboundedChannel	1	94.65 ms	4.724 ms	13.631 ms	8.76	68,648 B	0.065
ConcurrentQueue	2	73.82 ms	2.862 ms	8.438 ms	1.01	132,584 B	1.000
ConcurrentNativeQueue	2	47.05 ms	0.934 ms	1.887 ms	0.65	744 B	0.006
UnboundedChannel	2	81.40 ms	1.728 ms	5.041 ms	1.12	35,688 B	0.269
ConcurrentQueue	4	114.38 ms	2.389 ms	7.043 ms	1.00	133,048 B	1.000
ConcurrentNativeQueue	4	53.16 ms	0.984 ms	1.502 ms	0.47	1,208 B	0.009
UnboundedChannel	4	86.96 ms	1.726 ms	2.884 ms	0.76	19,320 B	0.145
ConcurrentQueue	8	136.94 ms	2.810 ms	8.240 ms	1.00	265,304 B	1.000
ConcurrentNativeQueue	8	58.95 ms	1.637 ms	4.723 ms	0.43	2,136 B	0.008
UnboundedChannel	8	81.98 ms	1.627 ms	4.481 ms	0.60	20,248 B	0.076

数据解读

吞吐量：

• 单生产者时，ConcurrentQueue<T> 更快（10.87ms vs 18.14ms）。这是预期的——单线程下没有竞争，ConcurrentQueue<T> 的 Interlocked.Increment 比 CAS 循环更高效，且其段数组在托管堆上具有更好的局部性。
• 从 2 个生产者开始，ConcurrentNativeQueue<T> 反超。在 4 个生产者时快 2.15 倍（53ms vs 114ms），8 个生产者时快 2.32 倍（59ms vs 137ms）。
• 随着生产者增加，ConcurrentQueue<T> 性能急剧下降（从 11ms 退化到 137ms），而 ConcurrentNativeQueue<T> 退化极为平缓（18ms → 59ms）。这体现了 MPSC 单消费者路径的优势：消费者无竞争。
• Channel<T> 在所有配置下都最慢，因为 TryWrite/TryRead 路径额外维护了异步通知状态机。

内存分配：

• ConcurrentQueue<T> 每轮分配 ~130-265KB（托管段数组）
• ConcurrentNativeQueue<T> 分配仅 488-2,136B（BenchmarkDotNet 统计的是托管分配；线程同步原语的开销）
• 实际队列数据分配为 0 字节托管内存——全部在 NativeMemory 上，GC 完全不感知

扩展性：

生产者数	ConcurrentQueue	ConcurrentNativeQueue	加速比
1	10.87 ms	18.14 ms	0.60x
2	73.82 ms	47.05 ms	1.57x
4	114.38 ms	53.16 ms	2.15x
8	136.94 ms	58.95 ms	2.32x

可以看出，ConcurrentNativeQueue<T> 的优势随生产者数量增加而扩大。这是因为：

1. 消费者路径始终是零竞争的本地读写
2. 入队 CAS 只作用于段内的 EnqueuePos，而非全局队列头/尾
3. 缓存行填充消除了生产者-消费者之间的 false sharing

基准测试代码

using BenchmarkDotNet.Attributes;
using System.Collections.Concurrent;
using System.Threading.Channels;
using ConcurrentNativeQueueLibrary;

namespace ConcurrentNativeQueueBenchmark;

/// <summary>
/// MPSC 多生产者单消费者吞吐量：NativeQueue vs ConcurrentQueue vs Channel&lt;T&gt;，
/// 变化生产者数量观察扩展性。
/// 使用专用 Thread + 三阶段同步（ready/start/done）消除线程池调度抖动。
/// </summary>
[MemoryDiagnoser]
public class MpscBenchmark
{
    [Params(1, 2, 4, 8)]
    public int ProducerCount;

    private const int TotalItems = 1_000_000;

    private ConcurrentNativeQueue<long> _nativeQueue;
    private ConcurrentQueue<long> _managedQueue = null!;
    private Channel<long> _channel = null!;

    [IterationSetup]
    public void Setup()
    {
        _nativeQueue.Dispose();
        _nativeQueue = new ConcurrentNativeQueue<long>();
        _managedQueue = new ConcurrentQueue<long>();
        _channel = Channel.CreateUnbounded<long>(new UnboundedChannelOptions { SingleReader = true });
    }

    [GlobalCleanup]
    public void Cleanup() => _nativeQueue.Dispose();

    [Benchmark(Baseline = true)]
    public int ConcurrentQueue()
    {
        int itemsPerProducer = TotalItems / ProducerCount;
        int totalItems = itemsPerProducer * ProducerCount;
        var ready = new CountdownEvent(ProducerCount);
        var start = new ManualResetEventSlim(false);
        var done = new CountdownEvent(ProducerCount);
        var q = _managedQueue;

        for (int p = 0; p < ProducerCount; p++)
        {
            int pid = p;
            new Thread(() =>
            {
                ready.Signal();
                start.Wait();
                for (int i = 0; i < itemsPerProducer; i++)
                    q.Enqueue((long)pid * itemsPerProducer + i);
                done.Signal();
            }) { IsBackground = true }.Start();
        }

        ready.Wait();
        start.Set();

        int consumed = 0;
        while (consumed < totalItems)
        {
            if (q.TryDequeue(out _))
                consumed++;
        }

        done.Wait();
        ready.Dispose();
        start.Dispose();
        done.Dispose();
        return consumed;
    }

    [Benchmark]
    public int ConcurrentNativeQueue()
    {
        int itemsPerProducer = TotalItems / ProducerCount;
        int totalItems = itemsPerProducer * ProducerCount;
        var ready = new CountdownEvent(ProducerCount);
        var start = new ManualResetEventSlim(false);
        var done = new CountdownEvent(ProducerCount);

        for (int p = 0; p < ProducerCount; p++)
        {
            int pid = p;
            new Thread(() =>
            {
                ready.Signal();
                start.Wait();
                for (int i = 0; i < itemsPerProducer; i++)
                    _nativeQueue.Enqueue((long)pid * itemsPerProducer + i);
                done.Signal();
            }) { IsBackground = true }.Start();
        }

        ready.Wait();
        start.Set();

        int consumed = 0;
        while (consumed < totalItems)
        {
            if (_nativeQueue.TryDequeue(out _))
                consumed++;
        }

        done.Wait();
        ready.Dispose();
        start.Dispose();
        done.Dispose();
        return consumed;
    }

    [Benchmark]
    public int UnboundedChannel()
    {
        int itemsPerProducer = TotalItems / ProducerCount;
        int totalItems = itemsPerProducer * ProducerCount;
        var ready = new CountdownEvent(ProducerCount);
        var start = new ManualResetEventSlim(false);
        var done = new CountdownEvent(ProducerCount);
        var writer = _channel.Writer;
        var reader = _channel.Reader;

        for (int p = 0; p < ProducerCount; p++)
        {
            int pid = p;
            new Thread(() =>
            {
                ready.Signal();
                start.Wait();
                for (int i = 0; i < itemsPerProducer; i++)
                    writer.TryWrite((long)pid * itemsPerProducer + i);
                done.Signal();
            }) { IsBackground = true }.Start();
        }

        ready.Wait();
        start.Set();

        int consumed = 0;
        while (consumed < totalItems)
        {
            if (reader.TryRead(out _))
                consumed++;
        }

        done.Wait();
        ready.Dispose();
        start.Dispose();
        done.Dispose();
        return consumed;
    }
}

免责声明：以上基准测试结果仅代表特定硬件、操作系统和 .NET 运行时版本下的表现。不同的 CPU 架构（核心数、缓存层级、NUMA 拓扑）、操作系统调度策略、.NET 版本以及后台负载等因素均可能显著影响实际性能。建议在目标部署环境中自行运行基准测试，以获取适用于实际场景的数据。

八、总结

ConcurrentNativeQueue<T> 不是 ConcurrentQueue<T> 的替代品，而是针对特定约束（MPSC + unmanaged + 零 GC）的专用数据结构。它的设计在三个层面做了权衡：

1. 并发模型层面：放弃 MPMC 通用性，换取单消费者的零原子操作出队路径
2. 内存管理层面：放弃 GC 自动管理的便利，换取零 GC 压力和确定性内存回收
3. 类型系统层面：放弃引用类型支持，换取 native 内存直接存储和 AOT 兼容性

如果你的场景命中了这些约束——游戏引擎、音频管线、高频交易、Native interop——ConcurrentNativeQueue<T> 可以提供显著的性能优势。否则，请使用 ConcurrentQueue<T>。

九、附录：什么是 MPSC 模型

MPSC（Multi-Producer, Single-Consumer）是一种常见的并发队列模型，指多个生产者线程同时向队列写入数据，但只有一个消费者线程从队列读取数据。

Producer 1 ──┐
Producer 2 ──┼──→ [ Queue ] ──→ Consumer（唯一）
Producer N ──┘

如图所示

与更通用的 MPMC（Multi-Producer, Multi-Consumer）模型相比，MPSC 的核心优势在于：消费者端不存在竞争。由于只有一个线程执行出队操作，出队路径可以完全避免原子操作和锁，退化为普通的本地内存读写，从而获得极低的延迟和极高的吞吐量。

MPSC 模型天然适配以下架构模式：

• 事件驱动主循环：多个子系统产生事件，主线程逐一消费（游戏引擎、UI 框架）
• 日志聚合：多个业务线程写日志，单个后台线程批量刷盘
• Actor 模型：每个 Actor 拥有一个 MPSC 邮箱，其他 Actor 发送消息，目标 Actor 单线程处理
• I/O 事件合并：多个网络/磁盘回调线程产生事件，单个事件循环线程统一处理

MPSC 的限制也很明确：如果你的场景需要多个消费者并发消费同一个队列，则应选择 MPMC 队列（如 ConcurrentQueue<T>）或分区方案。

十、获取源码

完整源码、基准测试、单元测试及示例程序均已开源，欢迎访问 GitHub 仓库：

https://github.com/VAllens/ConcurrentNativeQueue

---

免责声明：本技术文章及 ConcurrentNativeQueue<T> 的实现代码均由 Claude Opus 4.6 High Thinking 辅助撰写/生成。ConcurrentNativeQueue<T> 尚未经过生产环境验证，如需在实际项目中使用，请自行进行充分、严谨的测试与审查。