最新要闻
- 焦点关注:什么是牙周炎及症状_什么是牙周炎
- SE终于承认《Forspoken》失败:但仍有可取之处
- 世界短讯!广西:全面实现高速服务区充电桩100%全覆盖
- 环球动态:鲜嫩如蛋挞 多鲜岩烧乳酪吐司1斤14.9元
- 世界滚动:Win10极限精简版系统Tiny10更新:加入远程桌面等实用功能
- Intel中国特供i5-13490F闪电降价!这性价比 没治了
- 当前简讯:女子沙漠种树16年让县城免于消失 11年前美国预言翻车
- 每日短讯:掀起汽车降价狂潮之后 湖北再投放5亿元消费券
- 全球讯息:突破142万人!《CS:GO》同时在线数达10年来最高点
- 2.8K OLED高刷屏!华硕灵耀14 2023轻薄本图赏
- 女子试驾比亚迪汉撞树:碗口粗的树都弯了 A柱完好无损
- 环球关注:面向 DevOps 的 Kubernetes 最佳安全实践
- 明明不缺钱 为什么总有人喜欢在家收集垃圾?
- 大众ID.家族新能源汽车跟进降价 最高优惠4万元
- 每日头条!一小时发电量超七万千瓦时!首台国产F级50兆瓦重型燃气轮机下线
- 单机身10499元:佳能入门全画幅微单EOS R8正式上架
广告
手机
iphone11大小尺寸是多少?苹果iPhone11和iPhone13的区别是什么?
警方通报辅警执法直播中被撞飞:犯罪嫌疑人已投案
- iphone11大小尺寸是多少?苹果iPhone11和iPhone13的区别是什么?
- 警方通报辅警执法直播中被撞飞:犯罪嫌疑人已投案
- 男子被关545天申国赔:获赔18万多 驳回精神抚慰金
- 3天内26名本土感染者,辽宁确诊人数已超安徽
- 广西柳州一男子因纠纷杀害三人后自首
- 洱海坠机4名机组人员被批准为烈士 数千干部群众悼念
家电
看点:C#中定义自己的消费队列(下)
(资料图)
一 背景
在上一篇中我们介绍了一个关于使用C#中的Queue来定义自己的消费队列,这篇文章我将再次使用Queue来定义另外一种消费队列,这个队列中会使用到System.Threading.Timer
来定义一个10ms的Interval,和上一篇中产生数据一个个消费不同这篇文章中介绍的消费队列中消费定时器时间间隔内的所有待消费项,前面我们还是一样会通过源码来一步步讲述其内部原理,最后会通过几个单元测试来验证对应的使用。
二 源码
using System;using System.Collections.Generic;using System.Diagnostics;using System.Threading;namespace Pangea.Common.Utility.Buffer{ public class CustomConsumeQueue : IDisposable { #region Fields public const int INTERVAL_CONSUMING = 10;//单位:ms // used for dispose private bool _isDisposed = false; protected const bool FINALIZE_DISPOSING = false; protected const bool EXPLICIT_DISPOSING = true; private Timer _timer; private Queue _queue; private readonly object _lockObj = new object(); private readonly Action> _consumerQueueItemsAction;#if DEBUG // the counter for working thread numbers in current; public int _threadCounter = 0;#endif #endregion #region Ctor public CustomConsumeQueue(Action> consumerQueueItemsAction) { _queue = new Queue(); _timer = new Timer(new TimerCallback(BeginTakeQueueItems)); _timer.Change(Timeout.Infinite, Timeout.Infinite); _consumerQueueItemsAction = consumerQueueItemsAction; } ~CustomConsumeQueue() { Dispose(FINALIZE_DISPOSING); } #endregion #region Methods #region Public public bool Add(T item) { if (_isDisposed) return false; lock (_lockObj) { _queue.Enqueue(item); _timer.Change(INTERVAL_CONSUMING, Timeout.Infinite); } return true; } public int PendingToConsumeCount() { lock (_lockObj) { return _queue.Count; } } public void Dispose() { Dispose(EXPLICIT_DISPOSING); GC.SuppressFinalize(this); } public void Dispose(bool disposingMode) { if (_isDisposed) return; if (disposingMode == EXPLICIT_DISPOSING) { // release managed resource whne dispose by explicit } _timer?.Dispose(); _timer = null; _isDisposed = true; } #endregion #region Private private void BeginTakeQueueItems(object state) { ThreadPool.QueueUserWorkItem(state1 => {#if DEBUG Interlocked.Increment(ref _threadCounter);#endif try { T[] itemsArray = null; lock (_lockObj) { itemsArray = new T[_queue.Count]; _queue.CopyTo(itemsArray, 0); _queue.Clear(); if (_isDisposed == false) { // may throw a disposed exception ((Timer)state).Change(Timeout.Infinite, Timeout.Infinite); } } Trace.WriteLine($"[{DateTime.Now:HH-mm-ss fff}] Begin into consume procedure,QueueCount:{itemsArray.Length},Time:{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}"); //begin consumer queue items _consumerQueueItemsAction.Invoke(itemsArray); } catch (Exception ex) { // TODO Log exception, terminate current thread } finally {#if DEBUG Interlocked.Decrement(ref _threadCounter);#endif } }); } #endregion #endregion }}
2.1 原理讲解
- Add 的过程在Add的时候除了将待消费项添加到默认的_queue对象中以外就开始设置定时器触发的Interval为
INTERVAL_CONSUMING
,这个值默认设置为10ms,也就是当前定时器的触发间隔在添加消费项的时候会默认开启。 - 定时器消费过程定时器消费的时候会一次性将10ms的间隔内所有消费项一次性取到,然后一次性消费掉,另外需要注意的是,在定时器触发时间回调的过程中需要重新设置定时器的Interval为
Timeout.Infinite
,这样就能够将原来的定时器停掉,在下一次再次添加的时候重新设置默认的Interval,这样就能够一次往复交替进行,这个是消费过程的主要逻辑。 - 实现IDispose模式当前的CustomConsumeQueue显式实现IDispose的接口,代码中也都实现了对_isDisposed字段的判断,这个在使用的过程中需要特别注意。
三 测试
单元测试用例如下:
using NUnit.Framework;using Pangea.Common.Utility.Buffer;using System;using System.Collections.Generic;using System.Diagnostics;using System.Threading;namespace ACM.Framework.Test.Modules.Utils{ public class CustomConsumeQueueTest { CustomConsumeQueue _consumingQueue = null; [SetUp] public void Setup() { CustomConsumeQueue consumeQueue = new CustomConsumeQueue(StartConsuming); _consumingQueue = consumeQueue; } [Test] public void GeneralWritingTest() { int consumedCount = 0; CustomConsumeQueue consumeQueue = new CustomConsumeQueue(queueItems => { int currentConsumedCount = queueItems.Count; Console.WriteLine($"[{DateTime.Now:HH-mm-ss fff}] {currentConsumedCount} items has been consumed."); consumedCount += currentConsumedCount; }); int producedCount = new Random(Guid.NewGuid().GetHashCode()).Next(1, 100); AddItem(consumeQueue, producedCount); int loopTime = 1000; while (consumedCount != producedCount) { Thread.Sleep(100); loopTime -= 100; } Assert.IsTrue(consumeQueue.PendingToConsumeCount() == 0); Assert.IsTrue(consumeQueue._threadCounter == 0); Assert.IsTrue(consumedCount == producedCount); } [Test] public void TimerWritingTest() { int times = 20; int paralleCount = 5; int completedThreadCount = 0; for (int i = 0; i < paralleCount; i++) { int localIndex = i; ThreadPool.QueueUserWorkItem(obj => { while (times-- > 0) { AddItem(_consumingQueue, new Random(Guid.NewGuid().GetHashCode()).Next(1, 100)); Thread.Yield(); } Interlocked.Increment(ref completedThreadCount); Trace.WriteLine($"***************Begin run into parralle thread:{localIndex}************************"); }); } while (completedThreadCount != 5) { Thread.Sleep(100); } Thread.Sleep(1000); Assert.IsTrue(_consumingQueue.PendingToConsumeCount() == 0); Assert.IsTrue(_consumingQueue._threadCounter == 0); } [Test] public void AddAfterDisposeTest() { string item1 = GetRandomString(); bool result1 = _consumingQueue.Add(item1); Assert.IsTrue(result1); Thread.Sleep(1000); // Console.WriteLine will cost lots of time Assert.IsTrue(_consumingQueue.PendingToConsumeCount() == 0); Assert.IsTrue(_consumingQueue._threadCounter == 0, $"[{DateTime.Now:HH-mm-ss fff}] Thread numbers is not zero."); _consumingQueue.Dispose(); string item2 = GetRandomString(); bool result2 = _consumingQueue.Add(item2); Assert.IsFalse(result2); } private void StartConsuming(IList queueItems) { Console.WriteLine($"[{DateTime.Now:HH-mm-ss fff}] {queueItems.Count} items has been consumed."); } private void AddItem(CustomConsumeQueue queue, int numbers) { IList items = new List(); for (int i = 0; i < numbers; i++) { items.Add(GetRandomString()); } foreach (var item in items) { queue.Add(item); Thread.Sleep(new Random(Guid.NewGuid().GetHashCode()).Next(1, 50)); } } private string GetRandomString() { return Guid.NewGuid().ToString(); } }}
测试的单元用例主要包含下面的几个部分:
- GeneralWritingTest()实现单个线程增加PendingItem,然后实现消费的过程。
- TimerWritingTest()使用5个独立的线程增加PendingItem,在每个线程中随机产生消费项,然后判断消费过程是否正确。
- AddAfterDisposeTest() 主要测试代码实现Dispose的过程,在添加然后消费后中途在显式的Dispose当前的Queue,判断后续是否能够再继续进行添加。
关键词:
-
Codeforces Round #857 Div.1 1801 E F 题解
点我看题1801E Gasolineprices先来看这样一道题:一个长为n的字符串,每个字符都是一个英文小写字母。...
来源: Codeforces Round #857 Div.1 1801 E F 题解
打开MASA Blazor的正确姿势6:表单验证
看点:C#中定义自己的消费队列(下)
焦点关注:什么是牙周炎及症状_什么是牙周炎
SE终于承认《Forspoken》失败:但仍有可取之处
焦点速看:前端设计模式——适配器模式
世界短讯!广西:全面实现高速服务区充电桩100%全覆盖
环球动态:鲜嫩如蛋挞 多鲜岩烧乳酪吐司1斤14.9元
世界滚动:Win10极限精简版系统Tiny10更新:加入远程桌面等实用功能
Intel中国特供i5-13490F闪电降价!这性价比 没治了
计应212小组讨论junit成果
当前简讯:女子沙漠种树16年让县城免于消失 11年前美国预言翻车
每日短讯:掀起汽车降价狂潮之后 湖北再投放5亿元消费券
全球讯息:突破142万人!《CS:GO》同时在线数达10年来最高点
【世界时快讯】ELF 文件
环球快资讯丨表数据量大优化方案设计
2.8K OLED高刷屏!华硕灵耀14 2023轻薄本图赏
女子试驾比亚迪汉撞树:碗口粗的树都弯了 A柱完好无损
环球关注:面向 DevOps 的 Kubernetes 最佳安全实践
焦点速递!9.3.3输入的符号2
明明不缺钱 为什么总有人喜欢在家收集垃圾?
大众ID.家族新能源汽车跟进降价 最高优惠4万元
每日头条!一小时发电量超七万千瓦时!首台国产F级50兆瓦重型燃气轮机下线
Windows10免费激活专业版亲测有效无需安装软件,附:Windows10停止自动更新教程正解版
单机身10499元:佳能入门全画幅微单EOS R8正式上架
中国2月动力电池装车量:宁德时代、比亚迪拿下超7成市场份额
环球要闻:13日明早将现寒潮过程最低气温:上班记得多穿点
承诺捐款1100万未兑现被母校起诉 校友回应:自己会像罗永浩一样
寻味北京 | 文化活动·北京城市副中心 公共文化服务
最资讯丨Python 中连接MSSQL,MySQL,SQLite,Redis,ElasticSearch,Mongodb,PostgreSQL,Oracle,R
世界新动态:秒杀面试题!JS中this指向的理解和运用
环球精选!.NET、.NET Framework 和 .NET Core
在Linux中如何注销其他 SSH 用户
资讯推荐:补贴9万后12万买C6 博主揭雪铁龙套路玩的深!低配比高配还贵
1TB存储影像机皇!小米13 Ultra或5月发布:同步小米平板6/小米手环8
焦点速看:洛谷 P1015 回文数
世界滚动:曾称花40万买燃油车很悲催!智己CEO:智己LS7近一半车主来自BBA
世界实时:《你好李焕英》后二次合作:贾玲第二部导演作品有张小斐
全球看热讯:EntityFramworkCore7笔记
【天天时快讯】董明珠谈“35岁职场危机”:不理解 人们要到60岁才退休
宏任务&微处理
主龙骨图片_主龙骨
一次看过瘾!《流浪地球》系列两部电影下周连映:时长5小时
焦点热门:读Java性能权威指南(第2版)笔记14_垃圾回收A
环球热门:全球第15!小米13 Pro DXO音频140分:险胜iPhone 13 Pro Max
今日最新!有六百万人 沉迷在B站上看“抓小偷”
天天报道:不退休不会卖股票!董明珠:格力没实现6000亿营收目标 有遗憾
怎么利用异步设计提升系统性能?
【天天新要闻】500km续航纯电MPV!李想首曝网约车D1 Plus:项目被叫停太遗憾
全球观焦点:Go 面向对象
全球快播:Vue——Vue初始化【三】
吕钱浩:苹果2026年可能会推真全面屏iPhone 那个时候中兴已是第7代了
焦点快报!博主阿秋准备坐地铁游香港:晒证件自证是90后
环球快资讯:手机换号伤不起!当代人换手机号的成本有多高?
百事通!iPhone 17 Pro将是最完美iPhone!果粉还要等两年时间
苦是一种享受!董明珠回应仍做手机:格力一些部件必靠进口 国产质量不达标
冷水江市气象局发布大风蓝色预警【Ⅳ级/一般】
自称从清华退学考北大博主删除视频:此前收获50万点赞 快速涨粉
世界热议:PMP项目变更管理及变更流程总结
利用CSS使博客园图片自动居中,而文字保持居左
天天热文:量化交易基础 - 012 - 检验中的假设条件
全球新资讯:2999元国产显卡抢疯了 Steam游戏实测能玩 老黄旧将打造
当前快看:实拍成都6-7级大风来袭:狂风呼啸树折腰 行人雨伞秒被吹翻
邹磊委员:推进“风光水火储”多能互补综合能源供给体系建设
Log4j 配置
每日速递:flink入门-流处理
全球消息!汉诺塔问题——分而治之(引入递归,解决重复子问题)
世界球精选!Java基础入门-数组练习
国内已超越GPS 北斗卫星今年再发三颗星:下一代已在路上
世界观点:称根据身体来划分性别比较好 女星桥本爱被LGBT攻击
【独家】女生月薪两万辞职考研八次失败:次次铩羽而归
全球新资讯:第128篇:浏览器存储(cookie、webStorage、 IndexedDB)
世界简讯:关于使用python脚本将同级的其他目录下的所有文件根据年份移动到当脚本位置的年份目录
环球今热点:SSD比机械硬盘更容易坏?实测来了:跟想象中不一样
全球观天下!2月轿车销量排名:两“逸”同病相怜、秦汉一马平川
通讯!姚劲波代表:高学历干家政可能会越发普遍
《庆余年2》冲上热搜榜:网曝郭麒麟演的角色范思辙换人了
资讯:shiro-550反序列化漏洞分析
资讯:64.异常
世界热文:[新媒体运营]新媒体运营概述
9名初中生花18元点18串里脊庆生 网友看哭:回不去的年华
今日报丨金钱豹回拜游客 动物园回应:没人教、第一次见
世界今日报丨隅田川进口胶囊咖啡液8杯到手10.9元:更鲜更好喝
有赞一面:还有任务没执行,线程池被关闭怎么办?
全球快看点丨注解与反射
java基础二-面向对象的三大特性
全球最资讯丨A卡有能力跟RTX 4090正面刚 AMD:不做是因为太贵、功耗高
微头条丨深山红烛 照亮希望
环球新消息丨Redis 深度学习
环球快资讯:前端设计模式——策略模式
新消息丨MySQL学习笔记-SQL实践1
每日快报!安全性不行 Win7系统真的别用了 又一个游戏《堡垒之夜》放弃支持
每日快播:童年回来了!《灌篮高手》发布流川枫角色海报
每日报道:使用web client对 vcenter 进行补丁升级
有监督学习——梯度下降
全球百事通!聪明的燕姿
当前视点!Oracle数据库中没有scott账户的方法
今日热闻!光速破发?iPhone 14黄色版还未开售便降价600元
华硕发布ROG Strix Impact III鼠标:双手通用、可更换微动插槽
真香!重庆购买HUAWEI问界M7直补3万:起售仅25.98万