如果你写过多线程程序，大概率遇到过这类问题：一个模块拼命生产数据，另一个模块再怎么努力也处理不完，结果不是内存涨上天，就是某个任务卡死在那里。C#生产者-消费者模式就是专门解决这种场景的经典方案。这篇内容我打算从最朴素的原理讲起，再给出C#里几种常用实现，包括BlockingCollection、Channel，以及手写队列+信号量的做法，最后结合一个模拟日志采集场景走一遍完整代码，把容量设置、线程数选择、取消和异常处理这些细节都拆开聊清楚。无论你是刚接触并发编程，还是已经写过一段时间多线程但经常在边界条件上翻车，这篇都可以拿来做参考。
1. 为什么要用C#生产者-消费者模式解耦业务
1.1 先从生活里的排队场景说起
想象一下食堂打饭的窗口。厨师在后厨不停做菜，学生在窗口排队取餐。如果厨师做完一道菜就直接冲到窗口，学生涌上来一人一个窗口，场面一定失控。中间那个打餐台就是缓冲区，厨师是生产者，学生是消费者。缓冲区存在的意义是让两边不必直接对接，谁快谁慢都互不拖累。生产者和消费者模式的核心就是这个缓冲区。
写代码时也一样。你有一个定时采集数据的程序，又有一个批量写入数据库的服务。采集很快，写库很慢。如果采集完直接同步调用写库，采集线程就得等着数据库提交完成，吞吐量被严重拖慢。更麻烦的是，如果数据库偶尔抖动一下，采集线程也跟着一起阻塞，整个链路全部瘫痪。引入一个中间队列，生产方只要把数据丢进队列就立刻返回，消费方按自己的节奏从队列里取数据，两边各干各的，互不干扰。
1.2 模式的核心价值：解耦、削峰、异步化
解耦是这套模式最直观的好处。生产方不需要知道消费方有多少个、是怎么实现的，只需要面向队列写数据。消费方也不需要关心数据是来自网络请求还是文件扫描，只需要从队列里读。系统内部的模块边界一下子就清楚了。
削峰则是应对突发流量的关键。比如外卖订单系统在午高峰瞬间来了几千个订单，如果每个订单都立刻调支付接口、调库存接口，后端的压力会被直接打爆。用一个有容量上限的队列把请求先收进来，消费者按稳定速率慢慢处理，高峰就被“削”掉了。当然，队列只能暂时缓冲，如果峰值持续时间过长，队列被填满之后还是要面对限流，但至少不会一上来就崩。
异步化说的是从调用方视角看，生产操作能立刻返回，业务响应速度快了。用户点了个按钮，后台把任务扔进队列，界面马上提示“已受理”，而不是等十分钟后才告诉你成功。这在体验上的收益非常明显。
1.3 什么场景适合用，什么场景别硬套
适合用的场景有几类比较典型：一是任务提交速度不规律，时快时慢；二是任务处理耗时不稳定，比如依赖外部接口或文件IO；三是需要把较大的工作拆成多个并发消费者并行处理；四是想给系统加上优雅关停、任务排队、失败重试等能力。
不太适合的场景也有。如果生产方和消费方速度基本一致，而且都是纯内存计算，那加一层队列反而增加开销和复杂度。另外，如果系统已经用了消息中间件（比如某种分布式消息队列），你还需要区分一下：本地进程内的生产者-消费者模式解决的是进程内部线程之间的协作，消息中间件解决的是多个服务之间的数据传递，两者不是替代关系。某些情况下可以结合，但不要把同一个组件既当线程队列又当跨服务总线。
2. C#生产者-消费者模式的三种实现对比
2.1 BlockingCollection ：老牌稳重的同步方案
C#里最经典的生产者-消费者实现要数BlockingCollection 。它底层默认基于ConcurrentQueue ，并且在此基础上加上了阻塞和边界限制能力。生产者调用Add方法，如果队列已经达到最大容量，Add会阻塞，直到有消费者取走元素腾出空间。消费者用GetConsumingEnumerable或者TryTake来取数据，队列为空时就阻塞等待。
下面是最基础的一段用法。
var queue = new BlockingCollection<string>(boundedCapacity: 100); // 生产者 Task producer = Task.Run(() => { for (int i = 0; i < 1000; i++) { queue.Add($"data-{i}"); } queue.CompleteAdding(); // 标记生产完成 }); // 消费者 Task consumer = Task.Run(() => { foreach (var item in queue.GetConsumingEnumerable()) { Console.WriteLine($"处理: {item}"); } }); Task.WaitAll(producer, consumer);这套API写起来很顺手。BlockingCollection 内部已经帮你处理了线程安全、阻塞等待、集合本身的并发修改，不需要自己加锁。它的CompleteAdding方法尤其重要，只有生产方全部完成后调用了它，消费者侧的foreach循环才会正常退出，否则消费者会一直等下去。
不足之处也有。BlockingCollection是同步阻塞模型，在UI线程或者高并发服务中如果不小心，容易把线程池线程占住。它在Windows平台上能用，但在某些高性能场景下吞吐量不如后来出现的Channel 。
2.2 Channel ：高性能异步优先的现代选择
Channel 在.NET Core 3.0之后被正式引入，设计目标非常明确：为生产者-消费者模式提供异步、高性能、线程安全的管道。它把操作分成了Writer和Reader两个独立API，结构性地区分了读写两侧。
用Channel实现同样功能的代码如下。
var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.Wait, SingleReader = false, SingleWriter = false, AllowSynchronousContinuations = false }); // 生产者 async Task ProduceAsync() { for (int i = 0; i < 1000; i++) { await channel.Writer.WriteAsync($"data-{i}"); } channel.Writer.TryComplete(); // 标记完成 } // 消费者 async Task ConsumeAsync() { await foreach (var item in channel.Reader.ReadAllAsync()) { Console.WriteLine($"处理: {item}"); } } Task producerTask = ProduceAsync(); Task consumerTask = ConsumeAsync(); await Task.WhenAll(producerTask, consumerTask);Channel的一大优势是API天然适配async/await。生产方可以用WriteAsync异步等待写入而不会阻塞线程；消费方用ReadAllAsync拿到一个异步数据流，配合await foreach，写起来就跟同步遍历一样简单。它还有多种FullMode策略，包括Wait、DropNewest、DropOldest和DropWrite。容量满后到底让生产方等着，还是丢掉部分数据，完全由你业务决定。
2.3 手写队列+信号量：真正理解底层逻辑
有时候项目比较老，不能用新语法，或者你就是想用更细粒度控制实现，那可以基于ConcurrentQueue 配合SemaphoreSlim自己拼一个队列。ConcurrentQueue本身是线程安全的先进先出集合，但它缺少“空队列等待”和“满队列阻塞”这两个语义，信号量主要就是为了补齐这两个行为。
我见过一种比较简洁的手写方式：
public class SimpleQueue<T> { private readonly ConcurrentQueue<T> _queue = new(); private readonly SemaphoreSlim _notEmpty = new(0, int.MaxValue); private readonly SemaphoreSlim _notFull; private readonly int _capacity; public SimpleQueue(int capacity) { _capacity = capacity; _notFull = new SemaphoreSlim(capacity, capacity); } public async Task EnqueueAsync(T item, CancellationToken token = default) { await _notFull.WaitAsync(token); _queue.Enqueue(item); _notEmpty.Release(); } public async Task<T> DequeueAsync(CancellationToken token = default) { await _notEmpty.WaitAsync(token); bool ok = _queue.TryDequeue(out var item); if (ok) _notFull.Release(); return item; } }这个实现的核心逻辑是：入队前先等待“有空位”的信号量，入队后释放“队列非空”信号量；出队时则反过来。用两个信号量分别代表容量和元素数量，队列满和队列空的双方都会阻塞等待，既不会无锁空转，也不会覆盖数据。当然这套实现只是简化版，实际工程里你还需要处理Complete信号、异常回滚、动态扩容等问题。自己手写一次最大的好处是彻底弄懂阻塞是怎么回事，这对排查线上问题特别有帮助。
2.4 三种方式的选型建议
面对一个具体项目时，最忌讳的是“因为我会某个API就用某个API”。我一般按这样的思路做选择。
如果你的项目运行在.NET Core 3.0以上，没有特殊限制，优先用Channel 。它异步性能好，吞吐高，API语义清晰，可扩展性好。BlockingCollection 适合老项目升级成本低的情况，或者你在同步代码里已经大量使用了Task.Run，不想改造成async/await风格。手写队列一般只在研究学习、自定义调度、或者要兼容非常特殊的框架时才会碰，日常业务开发完全没必要重复造轮子。
一句话总结：默认选Channel，老项目维持BlockingCollection，想搞懂原理就手写一遍。
3. 实战：用Channel实现一个多生产者多消费者流水线
3.1 模拟日志采集场景
纸上谈兵讲再多，不如跑一个能落地的例子。我用模拟日志采集来演示。假设我们要做一个后台日志采集服务：生产者线程不断从多个日志源拉取日志行，消费者线程负责把日志解析、清洗，然后写入一个伪终端。真实项目里可能是写数据库，或者发到Kafka，这里用控制台输出代替，重点是看模式本身。
业务需求拆成三条：
- 有5个生产者，每个生产者模拟一个日志文件流。
- 有3个消费者，同时从共享队列里取日志做处理。
- 队列容量限制为1000，防止内存被瞬间打爆。
- 支持取消，服务停止时等待已有日志处理完再退出。
3.2 完整实现代码
先定义日志模型和模拟生产者方法。
using System.Threading.Channels; public record LogEntry(DateTime Timestamp, string Source, string Content, int Sequence); static async Task ProduceLogsAsync( ChannelWriter<LogEntry> writer, string source, int total, CancellationToken cancellationToken) { try { for (int i = 0; i < total; i++) { // 模拟读取日志文件，偶尔耗时 await Task.Delay(Random.Shared.Next(1, 5), cancellationToken); var entry = new LogEntry( DateTime.UtcNow, source, $"log line {i} from {source}", i); // 核心写入操作；队列满时会等待 await writer.WriteAsync(entry, cancellationToken); } } catch (OperationCanceledException) { Console.WriteLine($"生产者 {source} 已取消"); } }消费者侧如下。
static async Task ConsumeLogsAsync( ChannelReader<LogEntry> reader, int consumerId, CancellationToken cancellationToken) { // await foreach 会在队列完成且元素读完时自动退出 await foreach (var entry in reader.ReadAllAsync(cancellationToken)) { // 模拟解析和清洗，通常比生产慢得多 await Task.Delay(Random.Shared.Next(5, 15), cancellationToken); // 这里可以写数据库、写远端、写本地 Console.WriteLine($"[消费者{consumerId}] 处理: {entry.Source} #{entry.Sequence} - {entry.Content}"); } Console.WriteLine($"消费者 {consumerId} 完成"); }主流程把它们组合起来。
var channel = Channel.CreateBounded<LogEntry>( new BoundedChannelOptions(1000) { FullMode = BoundedChannelFullMode.Wait, SingleWriter = false, // 有多个生产者并发写 SingleReader = false // 有多个消费者并发读 }); using var cts = new CancellationTokenSource(); var producers = Enumerable.Range(0, 5) .Select(i => ProduceLogsAsync(channel.Writer, $"source-{i}", total: 200, cts.Token)) .ToArray(); var consumers = Enumerable.Range(0, 3) .Select(i => ConsumeLogsAsync(channel.Reader, i, cts.Token)) .ToArray(); await Task.WhenAll(producers); // 所有生产者完成后，必须调用 Complete 让消费者正常退出 channel.Writer.TryComplete(); await Task.WhenAll(consumers);这段代码最值得注意的地方有两点。第一，Producer们并发调用WriteAsync，所以创建Channel时SingleWriter必须设为false。第二，务必在等待所有生产者结束后调用TryComplete。否则消费者端的ReadAllAsync会一直阻塞，程序永远不会结束。这就是生产者-消费者模式最常见的“卡死”问题。
3.3 其他关键参数怎么设置
BoundedChannelOptions里还有几个常用参数，我逐个说一下。
AllowSynchronousContinuations这个参数容易被忽略。设为true时，写入操作可能会在写入者线程上直接触发消费者的读取回调，减少一次线程调度，但也会导致“生产者的最后一个操作意外执行了消费者逻辑”。大多数场景保持false就行，性能敏感型管道可以做压测对比以后再开。
Capacity设多少取决于你能接受多少内存。假设每条日志1KB，容量10000就是约10MB缓冲。如果消费者短暂阻塞30秒，在每秒生产1000条的速率下，你需要至少30000条缓冲容量。算一下再设，别拍脑袋。
FullMode如果用DropNewest或DropWrite，意味着在容量满时会悄悄丢数据。日志类场景可能允许丢，但交易、订单场景绝对不能丢。Wait是最安全也最容易理解的选择。
3.4 BlockingCollection版本对照
如果换成BlockingCollection，主流程差别不大，主要API变成Add和TryTake，生产者结束后调CompleteAdding。性能上，BlockingCollection在纯同步阻塞场景里已经够用，但遇到高吞吐任务时，建议用Channel再压测一遍。有一年我优化一个数据上报服务，从BlockingCollection换成Channel后，只是由于减少了不必要的线程恢复开销，吞吐量就提升了大约30%。这个数字不绝对，但值得你重视异步等待带来的收益。
4. C#生产者-消费者模式最容易踩的坑
4.1 队列容量的错觉：内存不是无限的
最常见的问题是设计时不设容量上限，直接Channel.CreateUnbounded ()。看起来操作简单、永远不会阻塞生产者，实际上隐患特别大。如果消费者长时间故障，未消费的数据会一直堆积在内存里。数据量小还好，一旦业务量涨起来，几分钟就能把Heap吃掉几个GB。
有次我维护一个数据同步服务，某个上游接口突然变慢，消费者处理速度掉到原来的十分之一。因为队列是无界的，生产者还在拼命往里塞，最后GC长期处于告警状态，整个服务被频繁的Full GC拖到接近假死。后来排查下来，队列里存了上百万条消息。
所以生产环境务必使用有界队列。哪怕你预估压力不大，也至少设一个几千或几万的上线，避免雪崩时把进程内存打满。
4.2 消费端异常导致管道静默终止
消费者在foreach或await foreach循环里如果抛了未捕获异常，整个消费Task会终止，但生产者可能还蒙在鼓里继续往队列里写。如果队列是有界的，很快队列就会被填满，生产者也跟着阻塞，然后整个系统像死锁一样卡住。
正确的做法是在循环体内包try/catch，把单条数据的处理异常隔离掉。如果某条数据确实处理不了，可以记录错误日志，或者把它放入一个“死信队列”稍后重试。不要让单点错误杀死整个消费者循环。
await foreach (var entry in reader.ReadAllAsync(ct)) { try { await ProcessEntryAsync(entry, ct); } catch (Exception ex) { // 记录错误，继续下一个，不要 break 或 throw Console.Error.WriteLine($"处理异常: {ex.Message}"); } }4.3 忘记Complete导致消费者永远等待
这种“卡死”隐蔽性很强。代码测试时如果只跑了少量数据，可能生产结束得很快，消费者还没发现异常；一旦数据变多，生产者提前退出了某个分支，没有调用Complete/CompleteAdding，消费者的阻塞读取就永远停不下来。
排查方法也比较直接：检查所有生产者Task退出路径上是否都会调用Complete，包括异常分支。try/finally是标准做法。
try { // 生产者写逻辑 } finally { writer.TryComplete(); }4.4 循环里的Thread.Sleep空转问题
有些同学习惯用while(true)加TryDequeue，取不到数据就Thread.Sleep(100)，觉得这样简单直接。短时间看问题不大，实际有两个副作用：一是无意义的线程切换和唤醒浪费CPU；二是响应延迟被固定成你Sleep的间隔。
如果用的是Channel，完全没必要循环轮询，直接ReadAllAsync等待即可。就算用BlockingCollection，也有阻塞版的Take方法。所以不要自己写空转循环。真需要控制处理节奏时，可以考虑用PeriodicTimer或ChannelReader.WaitToReadAsync来做带等待的读取。
4.5 消费者数量不是越多越好
开通20个消费者Task并行处理看起来很高效，但每个Task都在做CPU计算或者上下文切换。如果任务是IO型的，适度增加并发确实能提高吞吐。如果是CPU密集型，超过CPU核心数的并发反而会造成线程频繁切换，效率下降。
一般做法是先按CPU核心数起步，再通过环境变量或配置来调整。上线后用压测数据说话，不要凭感觉配置。比如日志清洗这种IO占比较大的，3到5个并发通常就够了；如果处理中还有网络等待，可以适当调高。
5. C#生产者-消费者模式性能调优与扩展思路
5.1 先测量，再调优
性能调优最忌讳“我觉得”。要测量吞吐量和延迟，最简单的方式是在消费者处理逻辑里统计每秒处理条数。可以加一个计数器，每处理1000条打一次日志。
var processed = 0; var sw = System.Diagnostics.Stopwatch.StartNew(); await foreach (var item in reader.ReadAllAsync(ct)) { // 处理... processed++; if (processed % 1000 == 0) { var elapsed = sw.Elapsed.TotalSeconds; Console.WriteLine($"吞吐: {processed / elapsed:F0} 条/秒"); } }这段代码放到消费者循环中即可。观察生产者任务完成之前和之后的不同阶段，能帮你判断瓶颈是生产侧还是消费侧。
5.2 批量消费减少单条开销
如果任务本身很小，单条处理的开销主要在线程调度和循环边界上，批量消费能明显降低开销。Channel没有直接提供批量读取API，但可以循环从队列里poll多条再一次性处理。
var batch = new List<LogEntry>(capacity: 100); while (await reader.WaitToReadAsync(ct)) { while (batch.Count < 100 && reader.TryRead(out var item)) { batch.Add(item); } // 批量处理 await ProcessBatchAsync(batch, ct); batch.Clear(); }WaitToReadAsync会等待到队列非空或Complete，然后TryRead把可读元素一次性尽量取出来。这个模式适合写库、写文件的场景，减少单条网络请求或IO次数，效果立竿见影。
5.3 利用IAsyncEnumerable做更复杂的管道
生产者-消费者模式并不一定只在两个模块之间用。你可以把多级处理串成一个管道，第一级生产者读取日志，第二级消费者做解析，再把解析结果作为下一级生产者的输入。Channel的Reader本身实现了IAsyncEnumerable ，所以可以无缝返回给下一级。
async IAsyncEnumerable<LogEntry> ParseLogs(ChannelReader<LogEntry> reader) { await foreach (var entry in reader.ReadAllAsync()) { // 做解析或清洗 yield return entry; } }这种方式让每一级处理都保持异步流式，不会因为队列中转丢失数据，也不会把整批数据都加载到内存里。代码的可读性也比自己造回调链好很多。
5.4 从单机管道走向分布式队列
当业务规模继续增长，单机内存队列已经不够用，比如多个服务实例需要共享任务，或者系统重启后任务不能丢，这时候就要引入外部消息中间件。生产者-消费者模式的思想还是不变的，只是缓冲区从进程内的Channel换成了分布式队列。
有一个常见的误区是：本地队列用得好，就想着自己封装一个Redis队列来替代消息中间件。这事在小规模下可行，但你需要处理消费者竞争、确认消费、故障重试、顺序保证、持久化等一系列问题，工作量会迅速膨胀。真实生产环境，本地队列和分布式队列经常是两层配合：本地接收请求后立即落库，再异步发包到消息中间件，Consumer从中间件拿任务继续处理。理解了C#生产者-消费者模式，你再看那些分布式框架的源码时会发现很多熟悉的影子。
5.5 结合背压思路控制整体流量
最后聊一个比较高级的话题：背压。队列一旦满了，生产方就该知道“下游处理不过来了”。Channel的Wait模式天然实现了背压：生产方WriteAsync会等待，相当于下游通过队列容量反向控制上游速率。这种机制对保护系统非常有用。
但背部还有一层：如果生产方本身在接收外部请求，不能一直等待导致请求堆积在更上游的网关卡住。这种情况下更好的策略是上层引入限流，比如当队列水位超过80%时，直接拒绝一部分新请求并返回“系统繁忙”。这比让所有请求都一直在内存里排队要安全得多。生产者-消费者模式不是银弹，它把压力转移到缓冲区，真正的系统容量还是得靠整体设计来保证。
6. 写在最后的实战心得
用C#生产者-消费者模式这几年，我最大的感觉是，代码本身不难写，难的是对“阻塞”和“边界”的理解。你真正掌握它的标志不在于敲出Channel还是BlockingCollection，而在于队列满的时候你清楚会发生什么，生产者抛异常的时候消费者会不会卡死，服务关停时还能不能优雅把数据处理完。建议大家拿一个真实场景的小模块，比如日志上报、任务队列、消息推送，从有界Channel起步跑一遍，再加上取消和异常处理，跑完这些坑基本就都见过了。之后再遇到并发协作类问题，心里会踏实很多。
