C#中的Channel

博客园 - 桂素伟

Semantic Kernel:开启MCP Semantic Kernel:Phi-4 mini的tools .NET10：解决json序列化时引用自己 .NET10：字符数字 Semantic Kernel：接入本地deepseek-r1:1.5b Semantic Kernel：接入azure中的deepseek-r1 .NET9中基于策略角色验证的包冲突 .NET9中使用Options Semantic Kernel:OpenAPI的Plugin Semantic Kernel:Phi-4试用 AI应用开发的浅见 Semantic Kernel:Process Semantic Kernel:新Agent代理 UnitsNet 库简介 .NET9里WinForm更新了什么 SemanticKernel系列，AI系列，SmartFill介绍视频系列更流畅的asp.net api的错误返回用.srt字幕文件生成.wav语音自制实时翻译小工具

C#中的Channel

桂素伟 · 2025-03-02 · via 博客园 - 桂素伟

　　在 .NET 的异步编程中，System.Threading.Channels 提供了一种强大的方式来处理生产者-消费者模式，尤其是当我们要在不同的任务或服务之间传递数据时。这篇文章我们就来聊聊 UnboundedChannelOptions 和 BoundedChannelOptions 这两个类，以及它们的使用场景和区别。

代码背景介绍

　　我们先看一段代码，这是一个 ASP.NET Core 项目，它创建了一个 Channel<VectorData> 来存放消息，并通过 VectorService 来处理这些消息。

builder.Services.AddSingleton<Channel<VectorData>>(_ =>
{
    //return Channel.CreateUnbounded<VectorData>(new UnboundedChannelOptions
    //{
    //    SingleReader = true,
    //    AllowSynchronousContinuations = false,
    //});
    return Channel.CreateBounded<VectorData>(new BoundedChannelOptions(10)
    {
        SingleReader = true,
        FullMode = BoundedChannelFullMode.Wait,
        AllowSynchronousContinuations = false,
    });
});

这里我们有两种选择：

UnboundedChannelOptions（无限容量通道）
BoundedChannelOptions（有限容量通道）

那么，这两者有什么区别呢？让我们深入解析一下。

UnboundedChannelOptions：无限容量通道

当你使用 Channel.CreateUnbounded<T>(new UnboundedChannelOptions {...}) 时，意味着创建了一个没有大小限制的通道。这意味着：

消息可以无限存入，不会因为容量问题被阻塞。
可能会导致内存占用过大，如果生产者速度远超消费者，数据会一直积累，可能会导致 OOM（内存溢出）。
适用于生产者和消费者速度匹配，或有足够资源支持的场景

关键属性解析：

SingleReader：是否只有一个消费者读取数据。设为 true，意味着不会有多个任务同时消费消息，提高读取效率。
AllowSynchronousContinuations：是否允许同步执行回调。如果设为 false，所有等待的任务都会被异步调度，避免线程被阻塞。

适用场景：

生产者和消费者处理速度接近，不容易导致消息堆积。
应用场景对内存不是特别敏感，允许暂时存放大量数据。
任务执行时间较短，不容易积累大量待处理数据。

BoundedChannelOptions：有限容量通道

相对来说，BoundedChannelOptions 是容量受限的通道，它在创建时就指定了最大容量，如代码中的 new BoundedChannelOptions(10)。

关键特性：

通道容量是有限的，超过 10 条消息时，生产者不能再写入（除非有消费者取走消息）。
可以控制通道满时的行为，通过 FullMode 指定：

BoundedChannelFullMode.Wait（默认）：生产者等待，直到有空间可写入。
BoundedChannelFullMode.DropOldest：丢弃最早的数据，保证新数据能进入。
BoundedChannelFullMode.DropNewest：丢弃最新的数据，让旧数据保留。
BoundedChannelFullMode.DropWrite：直接丢弃当前要写入的数据。

关键属性解析：

SingleReader：同 UnboundedChannelOptions，控制是否只有一个消费者读取数据。
AllowSynchronousContinuations：是否允许同步执行任务，防止不必要的线程切换。

适用场景：

生产者速度可能会远超消费者，需要限制消息堆积，避免内存占用过高。
需要对通道满时的行为进行控制，例如优先保留新数据，或让生产者等待。

Unbounded vs Bounded：该如何选择？

特性	UnboundedChannelOptions	BoundedChannelOptions
是否限制容量	❌ 无限制	✅ 受限（如 10 条消息）
适用于高吞吐	✅ 适合	🚫 可能会影响
可能导致内存占满	✅ 可能	🚫 不太可能
生产者快于消费者时	❌ 可能导致 OOM	✅ 可以限制
配置通道满时行为	🚫 无法配置	✅ 可配置 `FullMode`
适用场景	生产者、消费者速度匹配	生产者速度可能远超消费者

简单来说：

如果你不确定生产者和消费者速度是否匹配，或者数据量不可控，建议用 BoundedChannelOptions。
如果你确定不会有大量数据积压，或者消费能力能跟上，UnboundedChannelOptions 也可以。

代码实践：修改为 UnboundedChannelOptions

如果我们改用 UnboundedChannelOptions，代码如下：

builder.Services.AddSingleton<Channel<VectorData>>(_ =>
{
    return Channel.CreateUnbounded<VectorData>(new UnboundedChannelOptions
    {
        SingleReader = true,
        AllowSynchronousContinuations = false,
    });
});

这里就没有了 BoundedChannelOptions 里的 FullMode，因为它本身不会满。

但是如果生产者不断写入，消费者来不及消费，可能会导致内存占用越来越大。

结论

UnboundedChannelOptions 适用于任务处理速度均衡的情况，但如果生产者远超消费者，会有 OOM 风险。
BoundedChannelOptions 可以防止内存爆炸，适用于高吞吐的生产者-消费者模式。
BoundedChannelOptions 可以通过 FullMode 控制通道满时的行为，而 UnboundedChannelOptions 没有这个选项。

如果你的数据量不确定，或者可能出现生产者>消费者的情况，建议优先选择 BoundedChannelOptions，以避免潜在的性能问题。

希望这篇文章能帮你理解这两种 Channel 选项的区别，让你在编写高并发、异步任务时能更好地选择合适的方案！

完整代码：

using System.Threading.Channels;

var builder = WebApplication.CreateBuilder(args);


builder.Services.AddHostedService<VectorService>();
builder.Services.AddSingleton<Channel<VectorData>>(_ =>
{
    //return Channel.CreateUnbounded<VectorData>(new UnboundedChannelOptions
    //{
    //    SingleReader = true,
    //    AllowSynchronousContinuations = false,
    //});
    return Channel.CreateBounded<VectorData>(new BoundedChannelOptions(10)
    {
        SingleReader = true,
        FullMode = BoundedChannelFullMode.Wait,
        AllowSynchronousContinuations = false,
    });
});
var app = builder.Build();


app.MapGet("/vector", async (Channel<VectorData> channel) =>
{
    await channel.Writer.WriteAsync(new VectorData($"这里是用户内容类信息，{DateTime.Now.ToString("yyyyMMddHHmmssfff")}"));
    return Results.Ok();
});

app.Run();
public record VectorData(string content);
public class VectorService : BackgroundService
{
    private readonly Channel<VectorData> _channel;

    public VectorService(Channel<VectorData> channel)
    {
        _channel = channel;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (await _channel.Reader.WaitToReadAsync(stoppingToken))
        {
            var vectorData = await _channel.Reader.ReadAsync(stoppingToken);
            Console.WriteLine($"开始向量化处理：{vectorData.content}");
            await Task.Delay(3000, stoppingToken);
            Console.WriteLine($"向量化处理结束");
        }
    }
}

　　文章来源微信公众号

　　想要更快更方便的了解相关知识，可以关注微信公众号

****欢迎关注我的asp.net core系统课程****
《asp.net core精要讲解》 https://ke.qq.com/course/265696
《asp.net core 3.0》 https://ke.qq.com/course/437517
《asp.net core项目实战》 https://ke.qq.com/course/291868
《基于.net core微服务》 https://ke.qq.com/course/299524

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - 桂素伟

代码背景介绍

UnboundedChannelOptions：无限容量通道

关键属性解析：

适用场景：

BoundedChannelOptions：有限容量通道

关键特性：

关键属性解析：

适用场景：

Unbounded vs Bounded：该如何选择？

代码实践：修改为 UnboundedChannelOptions

结论