Task+ConcurrentQueue多线程编程
2024-09-27 09:24:59
队列(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
最新文章
- android之短信拦截器
- __attribute__ 你知多少?
- Syslog
- 视频播放用户行为记录之使用Cache的方法
- linux命令每日一练习-pwd,cd
- mysql创建表与索引
- 利用js+canvas实现的时钟效果图
- XAML中的Path
- 数论之高次同余方程(Baby Step Giant Step + 拓展BSGS)
- hdu 5534 Partial Tree 背包DP
- PAT-乙级-1025. 反转链表 (25)
- mysql安装常见问题(系统找不到指定的文件、发生系统错误 1067 进程意外终止)
- 关于Java____________Object类
- SQL 生日得到年龄
- pytorch bug: for step,data in enumerate(loader)+Connection reset by peer
- JavaScript-合同到期续约案例
- lombok-@Accessors注解
- pycharm破解版
- Android 生态消息推送平台介绍
- go包管理工具glide
热门文章
- 【LeetCode】71. Simplify Path 解题报告(Python)
- The 2015 China Collegiate Programming Contest -ccpc-c题-The Battle of Chibi(hdu5542)(树状数组,离散化)
- window11连接局域网共享失败处理办法
- Pikachu漏洞练习-SQL-inject(持续更新)
- 【环境搭建】安装pyQt5 在pycharm报This application failed to start because no Qt platform plugin could be initialized的问题
- OverFeat:Integrated Recognition, Localization and Detection using Convolutional Networks
- [数学]高数部分-Part I 极限与连续
- Capstone CS5263|DP转HDMI 4K60HZ转换芯片|CS5263芯片|替代PS176芯片
- BL8810最新版规格书|BL8810方案|USB 2.0读卡器控制器
- python 安装包时提示“unsupport command install”