C#高级：使用ConcurrentQueue做一个简易进程内通信的消息队列-开发者社区

文章目录

一、简介
二、使用场景
三、好处
四、代码
总结

一、简介

使用ConcurrentQueue实现线程安全的进程内消息队列，支持多生产者/消费者模式。

二、使用场景

多线程间数据交换、异步任务处理、日志缓冲等需要线程安全队列的场景。

三、好处

线程安全：内置锁机制，无需额外同步
高性能：无锁设计减少竞争
易用性强：实例绑定Key，简化队列管理

四、代码

usingSystem;usingSystem.Collections.Concurrent;usingSystem.Collections.Generic;usingSystem.Linq;usingSystem.Threading;usingSystem.Threading.Tasks;/// <summary>/// 简易队列管理器（实例绑定唯一Key，无需重复传参）/// </summary>/// <typeparam name="T">队列元素类型</typeparam>publicclassEasyQueue<T>:IDisposable{// 静态字典：全局管理所有Key对应的队列（保证Key唯一）privatestaticreadonlyConcurrentDictionary<string,ConcurrentQueue<T>>_globalQueueDict=new();// 实例专属Key（创建实例时指定，后续操作无需传参）publicstringQueueKey{get;}// 实例绑定的队列（避免每次从全局字典获取）privateConcurrentQueue<T>_boundQueue;/// <summary>/// 构造函数：注册Key并绑定专属队列（核心优化点）/// </summary>/// <param name="queueKey">队列唯一标识（注册Key）</param>/// <exception cref="ArgumentNullException">Key为空时抛出</exception>publicEasyQueue(stringqueueKey){if(string.IsNullOrWhiteSpace(queueKey))thrownewArgumentNullException(nameof(queueKey),"队列Key不能为空或空白");QueueKey=queueKey;// 注册Key：不存在则创建新队列，存在则绑定已有队列（线程安全）_boundQueue=_globalQueueDict.GetOrAdd(queueKey,_=>newConcurrentQueue<T>());}#region核心方法（无需传Key，直接操作实例绑定的队列）/// <summary>/// 单个元素入队（实例方法，无需传Key）/// </summary>/// <param name="item">要入队的元素</param>publicvoidEnqueue(Titem){_boundQueue.Enqueue(item);}/// <summary>/// 批量元素入队（实例方法，无需传Key）/// </summary>/// <param name="items">要入队的元素列表</param>/// <exception cref="ArgumentNullException">列表为空时抛出</exception>publicvoidEnqueue(List<T>items){if(items==null||items.Count==0)thrownewArgumentNullException(nameof(items),"入队元素列表不能为空或空列表");foreach(variteminitems){_boundQueue.Enqueue(item);}}/// <summary>/// 出队（实例方法，无需传Key）/// </summary>/// <param name="item">出队的元素（队列为空时返回default(T)）</param>/// <returns>是否出队成功（true=成功，false=队列为空）</returns>publicboolDequeue(outTitem){return_boundQueue.TryDequeue(outitem);}/// <summary>/// 查询队列所有元素（实例方法，无需传Key）/// </summary>/// <returns>队列元素列表（队列为空返回空List）</returns>publicList<T>GetQueueItems(){return_boundQueue.ToList();}#endregion#region扩展实用方法（实例级，无需传Key）/// <summary>/// 判断当前实例绑定的队列是否为空/// </summary>publicboolIsEmpty=>_boundQueue.IsEmpty;/// <summary>/// 获取当前队列的元素数量/// </summary>publicintCount=>_boundQueue.Count;/// <summary>/// 清空当前队列/// </summary>publicvoidClear(){while(_boundQueue.TryDequeue(out_)){}}/// <summary>/// 静态方法：移除全局字典中的指定Key（注销队列）/// </summary>/// <param name="queueKey">要注销的Key</param>/// <returns>是否注销成功</returns>publicstaticboolUnregisterQueue(stringqueueKey){if(string.IsNullOrWhiteSpace(queueKey))thrownewArgumentNullException(nameof(queueKey));return_globalQueueDict.TryRemove(queueKey,out_);}/// <summary>/// 静态方法：查询全局已注册的所有队列Key/// </summary>/// <returns>所有Key的列表</returns>publicstaticList<string>GetAllRegisteredKeys(){return_globalQueueDict.Keys.ToList();}#endregion#region资源释放privatebool_disposed=false;protectedvirtualvoidDispose(booldisposing){if(!_disposed){if(disposing){// 释放实例时清空绑定的队列（可选，根据业务需求调整）Clear();}_disposed=true;}}publicvoidDispose(){Dispose(true);GC.SuppressFinalize(this);}#endregion}classProgram{staticvoidMain(string[]args){// 定义队列KeyconststringtestKey="MultiThreadQueue";// ========== 线程1：生产数据（入队） ==========TaskproduceTask=Task.Run(()=>{// 创建生产端实例（绑定testKey）varproducerQueue=newEasyQueue<int>(testKey);for(inti=1;i<=100;i++){producerQueue.Enqueue(i);Console.WriteLine($"【生产线程】入队：{i}");Thread.Sleep(1000);// 模拟生产耗时}Console.WriteLine("【生产线程】生产完成！");});// ========== 线程2：消费数据（出队） ==========TaskconsumeTask=Task.Run(()=>{// 创建消费端实例（绑定同一个testKey）varconsumerQueue=newEasyQueue<int>(testKey);intconsumeCount=0;// 循环消费，直到生产完成且队列为空while(!produceTask.IsCompleted||!consumerQueue.IsEmpty){if(consumerQueue.Dequeue(outintitem)){consumeCount++;Console.WriteLine($"【消费线程】出队：{item}");}else{Thread.Sleep(5);// 队列为空时短暂等待，避免空轮询}}Console.WriteLine($"【消费线程】消费完成！共消费{consumeCount}个元素");});// 等待所有线程完成Task.WaitAll(produceTask,consumeTask);Console.WriteLine("\n所有操作完成！");}}