在 .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:该如何选择?
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($"向量化处理结束");
}
}
}