队列(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

最新文章

  1. android之短信拦截器
  2. __attribute__ 你知多少?
  3. Syslog
  4. 视频播放用户行为记录之使用Cache的方法
  5. linux命令每日一练习-pwd,cd
  6. mysql创建表与索引
  7. 利用js+canvas实现的时钟效果图
  8. XAML中的Path
  9. 数论之高次同余方程(Baby Step Giant Step + 拓展BSGS)
  10. hdu 5534 Partial Tree 背包DP
  11. PAT-乙级-1025. 反转链表 (25)
  12. mysql安装常见问题(系统找不到指定的文件、发生系统错误 1067 进程意外终止)
  13. 关于Java____________Object类
  14. SQL 生日得到年龄
  15. pytorch bug: for step,data in enumerate(loader)+Connection reset by peer
  16. JavaScript-合同到期续约案例
  17. lombok-@Accessors注解
  18. pycharm破解版
  19. Android 生态消息推送平台介绍
  20. go包管理工具glide

热门文章

  1. 【LeetCode】71. Simplify Path 解题报告(Python)
  2. The 2015 China Collegiate Programming Contest -ccpc-c题-The Battle of Chibi(hdu5542)(树状数组,离散化)
  3. window11连接局域网共享失败处理办法
  4. Pikachu漏洞练习-SQL-inject(持续更新)
  5. 【环境搭建】安装pyQt5 在pycharm报This application failed to start because no Qt platform plugin could be initialized的问题
  6. OverFeat:Integrated Recognition, Localization and Detection using Convolutional Networks
  7. [数学]高数部分-Part I 极限与连续
  8. Capstone CS5263|DP转HDMI 4K60HZ转换芯片|CS5263芯片|替代PS176芯片
  9. BL8810最新版规格书|BL8810方案|USB 2.0读卡器控制器
  10. python 安装包时提示“unsupport command install”