Task+ConcurrentQueue多线程编程

  • A+
所属分类：.NET技术
摘要

   　队列（Queue）代表了一个先进先出的对象集合。当您需要对各项进行先进先出的访问时，则使用队列。当您在列表中添加一项，称为入队，当您从列表中移除一项时，称为出队。
便宜好用的国外VPS推荐

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

 

便宜好用的国外VPS推荐