首选 BlockingCollection，因其底层封装线程安全集合与阻塞/取消语义，自动处理取空阻塞、满阻塞和取消通知，支持有界缓冲、自动完成信号及统一异常处理。
直接用 BlockingCollection<t></t>，别手写锁和队列，它就是为这场景设计的。 手动用 Monitor、AutoResetEvent 或 ConcurrentQueue<t></t> + 自旋等待，容易出死锁、假唤醒、吞异常或资源泄漏。
为什么首选 BlockingCollection<t></t>
它底层封装了线程安全的集合（默认 ConcurrentQueue<t></t>）+ 阻塞/取消语义，自动处理“取空时阻塞”“满时阻塞”“取消通知”三件事。你不用管 WaitOne() 是否被中断、PulseAll() 是否漏通知、消费者是否在 Take() 前就被 CancellationToken 中止。
-
BlockingCollection<t></t>构造时可传入boundedCapacity实现有界缓冲区，避免内存无限增长 - 调用
CompleteAdding()后，后续GetConsumingEnumerable()会自然退出循环，无需额外标志位 - 所有操作（
Add()、Take()、TryAdd()）都响应CancellationToken，异常类型统一为OperationCanceledException
GetConsumingEnumerable() 是最简消费写法
别再写 while (true) { try { item = collection.Take(); } catch ... }。用 GetConsumingEnumerable() 让 foreach 自动处理完成信号和取消：
var collection = new BlockingCollection<string>(boundedCapacity: 100);
// 生产者
Task.Run(() => {
for (int i = 0; i < 1000; i++) {
collection.Add($"item-{i}");
Thread.Sleep(10);
}
collection.CompleteAdding(); // 必须调用，否则 foreach 不会结束
});
// 消费者（推荐）
foreach (var item in collection.GetConsumingEnumerable()) {
Console.WriteLine($"Consumed: {item}");
Thread.Sleep(5);
}
- 如果没调用
CompleteAdding()，GetConsumingEnumerable()会永远等待新项，程序无法退出 - 该枚举器内部已处理
CancellationToken，若需取消，请传入构造时指定的 token（通过new BlockingCollection<t>(collection, token)</t>） - 不支持 LINQ 方法如
.Where()或.Skip()，它是只读流式迭代器
多消费者共用一个 BlockingCollection 时要注意什么
多个 foreach 循环同时遍历同一个 GetConsumingEnumerable() 实例？不行——每个消费者必须调用自己的 GetConsumingEnumerable()，否则会共享迭代状态导致漏项或异常。
- 正确做法：每个消费者线程独立调用
collection.GetConsumingEnumerable() - 错误写法：
var enu = collection.GetConsumingEnumerable(); Task.Run(() => foreach(var x in enu) {...}); Task.Run(() => foreach(var x in enu) {...});—— 第二个 foreach 立即结束 - 若需精确控制并发数（比如最多 3 个消费者），用
Parallel.ForEach(collection.GetConsumingEnumerable(), ...)并设MaxDegreeOfParallelism，但注意它不保证顺序且异常会聚合抛出
真正难的是边界情况：生产者中途崩溃未调用 CompleteAdding()、消费者处理逻辑抛异常未捕获导致整个 foreach 退出、或 boundedCapacity 设太小造成生产者频繁阻塞影响上游吞吐。这些没法靠模式本身解决，得靠超时重试、外部健康检查和日志埋点来兜底。
