何时应优先选用System.Threading.Channels而非ConcurrentQueue？

测试背景与结果 #

我基于ConcurrentQueue<T>和SemaphoreSlim实现了一套生产者/消费者系统，随后用System.Threading.Channel实现了替代版本。通过BenchmarkDotNet进行基准测试：向两个系统各写入1000条数据，重复1000次并等待读取完成，结果如下：

可以看到ConcurrentQueue的实现速度明显快于Channel。我尝试将Channel的SingleReader和SingleWriter设置为true，但性能反而更差：

我不确定是实现代码或基准测试存在缺陷，还是结果真实有效。若结果有效，何时应优先选用Channels而非普通的ConcurrentQueue？

实现代码 #

MyQueue（ConcurrentQueue+SemaphoreSlim实现） #

public class MyQueue
{
    ConcurrentQueue<Item> _queue;
    SemaphoreSlim _readerFinishedSemaphore;
    SemaphoreSlim _readSemaphore;

    bool completed = false;

    public void Setup()
    {
        _queue = new();
        _readerFinishedSemaphore = new(0);
        _readSemaphore = new(0);

        var task = new Task(Reader, TaskCreationOptions.LongRunning);
        task.Start();
    }

    private async void Reader()
    {
        while (true)
        {
            await _readSemaphore.WaitAsync();
            while (_queue.TryDequeue(out var item))
            {
                // do stuff ...
            }

            if (completed) break;
        }

        _readerFinishedSemaphore.Release();
    }

    public void Write(IList<Item> items)
    {
        foreach (var i in items)
        {
            _queue.Enqueue(i);
        }

        _readSemaphore.Release();
    }

    public void CompleteAndWaitForReader()
    {
        completed = true;
        _readSemaphore.Release();
        _readerFinishedSemaphore.Wait();
    }
}

MyChannel（System.Threading.Channel实现） #

public class MyChannel
{
    Channel<Item> _channel = null!;
    SemaphoreSlim _readerFinishedSemaphore = null!;

    public void Setup()
    {
        _readerFinishedSemaphore = new(0);
        _channel = Channel.CreateUnbounded<Item>();

        var task = new Task(Reader, TaskCreationOptions.LongRunning);
        task.Start();
    }

    private async void Reader()
    {
        var reader = _channel.Reader;

        while (await reader.WaitToReadAsync())
        {
            while (reader.TryRead(out var item))
            {
                // do stuff ...
            }
        }

        _readerFinishedSemaphore.Release();
    }

    public void Write(IList<Item> items)
    {
        foreach (var i in items)
        {
            _channel.Writer.TryWrite(i);
        }
    }

    public void CompleteAndWaitForReader()
    {
        _channel.Writer.Complete();
        _readerFinishedSemaphore.Wait();
    }
}

基准测试代码 #

// items are generated in [GlobalSetup] using fixed-seed Random class

[IterationSetup]
public void IterationSetup()
{
    myChannel = new MyChannel();
    myQueue = new MyQueue();

    myChannel.Setup();
    myQueue.Setup();
}

[Benchmark]
public void MyQueue()
{
    for (int i = 0; i < Iterations; i++)
        myQueue.Write(items);

    myQueue.CompleteAndWaitForReader();
}

// same for MyChannel

测试运行环境为.NET 8.0.0-preview.6.23329.4。

分析与结论 #

结果有效性分析 #

你的测试结果在当前场景下是真实的，核心差异来自两个实现的同步逻辑：

• MyQueue的Reader通过SemaphoreSlim手动控制唤醒时机，每次批量写入后仅触发一次唤醒，Reader一次性处理所有累积的队列项，减少了异步等待和上下文切换的开销。
• MyChannel的Reader依赖WaitToReadAsync()的异步唤醒机制，每次TryWrite都可能触发唤醒信号，导致Reader多次进入循环处理，增加了同步和调度的额外开销。
• 开启SingleReader/SingleWriter后性能下降，是因为Channel针对单读写的优化更适合持续流式读写的场景，而你的批量写入+单次批量处理模式无法发挥其优势，反而因内部同步路径的调整增加了额外开销。

何时优先选用Channel #

尽管在你的批量场景下ConcurrentQueue表现更好，但Channel在以下场景中更具优势：

• 异步流处理：需要与async/await深度集成，或返回IAsyncEnumerable进行异步遍历时，Channel的内置异步API会比手动管理Semaphore更简洁可靠。
• 背压控制：当需要限制队列长度防止内存溢出时，Channel.CreateBounded()可轻松实现自动背压，无需手动编写数量控制逻辑。
• 多生产者/多消费者协调：Channel内置了对多读写场景的支持，无需手动处理信号量的释放和同步逻辑，降低了出错概率。
• 复杂流水线场景：在ASP.NET Core请求处理、事件驱动架构等场景中，Channel可以更自然地融入异步生态，实现组件间的解耦和流式数据传递。

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