寻求Channel/BlockingCollection的无内存分配替代方案及实现方法

问题根源 #

ChannelReader.ReadAsync和BlockingCollection.TryTake在热路径中产生分配的原因：

• Channel的异步方法即使返回ValueTask，当队列空时仍会分配状态机或等待Task；通用Channel为支持多生产者/多消费者，内部同步逻辑带来额外开销。
• BlockingCollection依赖的ConcurrentQueue存在内部节点分配，同步机制也有隐性内存开销，无法完全避免分配。

针对单生产者/单消费者场景，我们可以利用特性优化，实现完全无分配的队列操作。

优化现有Channel用法（减少分配） #

如果不想自定义队列，可先通过配置Channel的单生产者/单消费者选项，配合WaitToReadAsync+TryRead模式大幅降低分配：

// 创建单生产者单消费者的无界Channel
var channel = Channel.CreateUnbounded<JobMeta>(new UnboundedChannelOptions
{
    SingleReader = true,
    SingleWriter = true
});
var reader = channel.Reader;

while (!token.IsCancellationRequested)
{
    // WaitToReadAsync在队列有数据时返回已完成的ValueTask，无分配
    if (await reader.WaitToReadAsync(token))
    {
        // TryRead完全无分配，批量读取所有可用数据
        while (reader.TryRead(out var jobMeta))
        {
            jobMeta.Job.Execute();
            jobMeta.JobHandle.Notify();
        }
    }
}

这种方式在队列非空时完全无分配，仅在队列空等待时可能产生少量分配，适合大部分时间队列有数据的热路径。

自定义无分配单生产者单消费者队列（彻底解决） #

利用环形缓冲区（Ring Buffer）实现无锁、无分配的队列，完全适配单生产者/单消费者场景：

核心实现 #

public sealed class SingleProducerSingleConsumerQueue<T>
{
    private readonly T[] _buffer;
    private readonly int _mask;
    private volatile int _writeIndex;
    private volatile int _readIndex;
    private readonly ManualResetEventSlim _dataAvailableEvent = new ManualResetEventSlim(false);

    public SingleProducerSingleConsumerQueue(int initialCapacity = 1024)
    {
        // 确保容量为2的幂，用位运算替代取模，提升性能
        initialCapacity = initialCapacity < 2 ? 2 : NextPowerOfTwo(initialCapacity);
        _buffer = new T[initialCapacity];
        _mask = initialCapacity - 1;
    }

    private static int NextPowerOfTwo(int value)
    {
        value--;
        value |= value >> 1;
        value |= value >> 2;
        value |= value >> 4;
        value |= value >> 8;
        value |= value >> 16;
        return value + 1;
    }

    // 生产者写入，无内存分配
    public bool Enqueue(T item)
    {
        var currentWrite = Volatile.Read(ref _writeIndex);
        var nextWrite = (currentWrite + 1) & _mask;

        // 队列满，无法写入
        if (nextWrite == Volatile.Read(ref _readIndex))
            return false;

        _buffer[currentWrite] = item;
        Volatile.Write(ref _writeIndex, nextWrite);
        _dataAvailableEvent.Set();
        return true;
    }

    // 消费者读取，无内存分配
    public bool TryDequeue(out T item)
    {
        var currentRead = Volatile.Read(ref _readIndex);
        if (currentRead == Volatile.Read(ref _writeIndex))
        {
            item = default;
            _dataAvailableEvent.Reset();
            return false;
        }

        item = _buffer[currentRead];
        // 清空引用类型，避免内存泄漏
        if (typeof(T).IsReferenceType)
            _buffer[currentRead] = default;

        Volatile.Write(ref _readIndex, (currentRead + 1) & _mask);
        return true;
    }

    // 异步等待数据，无额外分配（复用事件对象）
    public async ValueTask<bool> WaitForDataAsync(CancellationToken token)
    {
        if (TryDequeue(out _))
            return true;

        try
        {
            await _dataAvailableEvent.WaitAsync(token);
            return TryDequeue(out _);
        }
        finally
        {
            if (Volatile.Read(ref _readIndex) == Volatile.Read(ref _writeIndex))
                _dataAvailableEvent.Reset();
        }
    }
}

使用示例 #

var queue = new SingleProducerSingleConsumerQueue<JobMeta>();

// 消费者线程/任务
while (!token.IsCancellationRequested)
{
    // 异步等待数据，无分配
    if (await queue.WaitForDataAsync(token))
    {
        // 批量读取所有可用数据，完全无分配
        while (queue.TryDequeue(out var jobMeta))
        {
            jobMeta.Job.Execute();
            jobMeta.JobHandle.Notify();
        }
    }
}

// 生产者写入
queue.Enqueue(new JobMeta { Job = myJob, JobHandle = handle });

关键特性 #

• Enqueue和TryDequeue方法完全无内存分配，依赖数组复用和位运算实现。
• 异步等待通过ManualResetEventSlim实现，复用内部事件对象，避免频繁创建Task。
• 环形缓冲区的容量自动调整为2的幂，最大化读写性能。

内容的提问来源于stack exchange，提问作者genaray