Task+ConcurrentQueue多线程编程
2022/2/3 12:12:26
本文主要是介绍Task+ConcurrentQueue多线程编程,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
队列(Queue)代表了一个先进先出的对象集合。当您需要对各项进行先进先出的访问时,则使用队列。当您在列表中添加一项,称为入队,当您从列表中移除一项时,称为出队。
ConcurrentQueue<T>队列是一个高效的线程安全的队列,是.Net Framework 4.0,System.Collections.Concurrent命名空间下的一个数据结构。
Task是在ThreadPool的基础上推出的,我们知道了ThreadPool的弊端:我们不能控制线程池中线程的执行顺序,也不能获取线程池内线程取消/异常/完成的通知。net4.0在ThreadPool的基础上推出了Task,Task拥有线程池的优点,同时也解决了使用线程池不易控制的弊端。
下面我们来看下Task+ConcurrentQueue实现多线程编程
1、定义最大线程数, 一般和本机的cpu 有关
/// <summary> /// 线程总数 /// </summary> private int threadNum = Convert.ToInt32(ConfigurationManager.AppSettings["ThreadNum"]);
ConcurrentQueue队列
/// <summary> /// 队列 /// </summary> private ConcurrentQueue<AssetRepayment> queues = new ConcurrentQueue<AssetRepayment>();
2、接下来 我们把业务数据取出来,加到定义的 queues.Enqueue(l);
var dt = DateTime.Now.Date; var list = AssetRepayService.GetRepayments().Where(o => o.AssetRepayStatus == AssetRepayStatus.NoSend && o.PlanRepaymentDate == dt && o.AssetRepayMode == AssetRepayMode.Withholding); int count = list.Count(); if (count == 0) { LogHelper.WriteFatal("代扣充值-没有可执行的数据"); return; } totalCount = count; int allpage = count / 200 + (count % 200 == 0 ? 0 : 1); int page = 0; LogHelper.WriteFatal("代扣充值-可执行的数据:" + count + "条,页数:"+ allpage); do { LogHelper.WriteFatal("代扣充值-第:" + page + "页"); var ll = list.OrderBy(o=>o.Id).Skip(page++ * 200).Take(200).ToList(); foreach (var l in ll) { queues.Enqueue(l); } } while (page < allpage);
3、等数据全部加载到queues,我们接着下一步
List<Task> tasks = new List<Task>(); for (int i = 0; i < threadNum; i++) { var task = Task.Run(() => { Process(); }); tasks.Add(task); } var taskList = Task.Factory.ContinueWhenAll(tasks.ToArray(), (ts) => { }); taskList.Wait();
利用Task 处理数据
这里需要注意的是
var taskList = Task.Factory.ContinueWhenAll(tasks.ToArray(), (ts) => { }); taskList.Wait();
这个代表开始执行线程并且需要全部完成 才会退出 ContinueWhenAll
4 接下去我们看下处理程序怎么消费数据
从队列取数据
var currentIndex = Interlocked.Increment(ref index); AssetRepayment repayId = null; var isExit = queues.TryDequeue(out repayId); if (!isExit) { break; }
有数据的话 往下走 执行我们要的业务逻辑
var service = context.GetService<IDeTransactionService>(); service.SubDeTransaction(repayId); LogHelper.WriteFatal(string.Format("代扣充值 共{0}条 当前第{1}条", totalCount, currentIndex));
完整处理方法如下
private void Process() { using (var context = new MefContext()) { while (true) { var currentIndex = Interlocked.Increment(ref index); AssetRepayment repayId = null; var isExit = queues.TryDequeue(out repayId); if (!isExit) { break; } try { var service = context.GetService<IDeTransactionService>(); service.SubDeTransaction(repayId); LogHelper.WriteFatal(string.Format("代扣充值 共{0}条 当前第{1}条", totalCount, currentIndex)); } catch (Exception ex) { LogHelper.WriteError("代扣充值-", ex); } } } }
到此为止,我们实现了 Task+ConcurrentQueue多线程编程。
完整代码块
链接:https://pan.baidu.com/s/1jgpafTFssiVLmZhDe1CgYQ
提取码:erib
这篇关于Task+ConcurrentQueue多线程编程的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-11-23增量更新怎么做?-icode9专业技术文章分享
- 2024-11-23压缩包加密方案有哪些?-icode9专业技术文章分享
- 2024-11-23用shell怎么写一个开机时自动同步远程仓库的代码?-icode9专业技术文章分享
- 2024-11-23webman可以同步自己的仓库吗?-icode9专业技术文章分享
- 2024-11-23在 Webman 中怎么判断是否有某命令进程正在运行?-icode9专业技术文章分享
- 2024-11-23如何重置new Swiper?-icode9专业技术文章分享
- 2024-11-23oss直传有什么好处?-icode9专业技术文章分享
- 2024-11-23如何将oss直传封装成一个组件在其他页面调用时都可以使用?-icode9专业技术文章分享
- 2024-11-23怎么使用laravel 11在代码里获取路由列表?-icode9专业技术文章分享
- 2024-11-22怎么实现ansible playbook 备份代码中命名包含时间戳功能?-icode9专业技术文章分享