一、方案对比：不同线程安全集合的适用场景

二、推荐方案及示例代码

方案 1：使用 BlockingCollection（同步模型）

public class QueueDemo
{private readonly BlockingCollection<int> _blockingCollection = new BlockingCollection<int>();private readonly CancellationTokenSource _cts = new CancellationTokenSource();public QueueDemo(){}// 生产者方法public void ProduceData(){Task.Run(() =>{var rnd = new Random();while (!_cts.IsCancellationRequested){var item = rnd.Next(1, 100);_blockingCollection.Add(item);      // 触发消费者唤醒Console.WriteLine($"Produced1: {item}");Thread.Sleep(500); // 模拟生产间隔//if(DateTime.Now > Convert.ToDateTime("2025-02-05 16:28:00")) break;}_blockingCollection.CompleteAdding(); // 结束消费});}// 消费者方法public void ConsumeData(){// 方式1：阻塞消费（推荐）Task.Run(() =>{try{Thread.Sleep(1000);// 使用阻塞方式消费（自动处理空队列等待）foreach (var item in _blockingCollection.GetConsumingEnumerable(_cts.Token)){// 自动等待新数据Console.WriteLine($"Consumed from BlockingCollection: {item}, 当前个数:{_blockingCollection.Count}");}}catch (OperationCanceledException){Console.WriteLine("Consumption canceled");}});}// 停止所有操作public void Stop(){_cts.Cancel();}
}//使用示例
var demo = new QueueDemo();
demo.ProduceData();
demo.ConsumeData();Console.WriteLine("Press any key to stop...");
Console.ReadKey();demo.Stop();

方案 2：使用 Channel（异步模型 - 推荐）

public class ChannelDemo
{private readonly Channel<int> _channel = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleWriter = false, SingleReader = false });private readonly CancellationTokenSource _cts = new CancellationTokenSource();// 生产者（异步写入）public async Task ProduceAsync(){while (true){var item = GenerateItem();await _channel.Writer.WriteAsync(item); // 非阻塞写入Console.WriteLine($"Produce: {item}");await Task.Delay(20);}}// 消费者（异步读取）public async Task ConsumeAsync(){while (await _channel.Reader.WaitToReadAsync()){if (_channel.Reader.TryRead(out var item)){await ProcessItemAsync(item);}}}private int GenerateItem() => new Random().Next(1, 100);private async Task ProcessItemAsync(int item){await Task.Delay(100); // 模拟异步处理Console.WriteLine($"Processed: {item}");}// 停止所有操作public void Stop(){_cts.Cancel();}
}

三、选型建议