快捷導(dǎo)航

C# 并行和多線程編程——并行集合和PLinq

更新時(shí)間：2021年02月20日 09:21:06 作者：雲(yún)霏霏

這篇文章主要介紹了C# 并行和多線程編程的相關(guān)資料,幫助大家更好的理解和學(xué)習(xí)使用c#，感興趣的朋友可以了解下

　　在上一篇博客，我們學(xué)習(xí)了Parallel的用法。并行編程，本質(zhì)上是多線程的編程，那么當(dāng)多個(gè)線程同時(shí)處理一個(gè)任務(wù)的時(shí)候，必然會(huì)出現(xiàn)資源訪問問題，及所謂的線程安全。就像現(xiàn)實(shí)中，我們開發(fā)項(xiàng)目，就是一個(gè)并行的例子，把不同的模塊分給不同的人，同時(shí)進(jìn)行，才能在短的時(shí)間內(nèi)做出大的項(xiàng)目。如果大家都只管自己寫自己的代碼，寫完后發(fā)現(xiàn)合并不到一起，那么這種并行就沒有了意義。
　　并行算法的出現(xiàn)，隨之而產(chǎn)生的也就有了并行集合，及線程安全集合；微軟向的也算周到，沒有忘記linq，也推出了linq的并行版本，plinq - Parallel Linq.

一、并行集合 —— 線程安全集合

　　并行計(jì)算使用的多個(gè)線程同時(shí)進(jìn)行計(jì)算，所以要控制每個(gè)線程對(duì)資源的訪問，我們先來看一下平時(shí)常用的List<T>集合，在并行計(jì)算下的表現(xiàn)，新建一個(gè)控制臺(tái)應(yīng)用程序，添加一個(gè)PEnumerable類(當(dāng)然你也直接寫到main方法里面測(cè)試，建議分開寫)，寫如下方法：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Collections.Concurrent;

namespace ThreadPool
{
 public class PEnumerable
 {
  public static void ListWithParallel()
  {
   List<int> list = new List<int>();
   Parallel.For(0, 10000, item =>
   {
   list.Add(item);
   });
   Console.WriteLine("List's count is {0}",list.Count());
  }
 }
}

點(diǎn)擊F5運(yùn)行，得到如下結(jié)果：

看到結(jié)果中顯示的5851，但是我們循環(huán)的是10000次啊！怎么結(jié)果不對(duì)呢？這是因?yàn)長(zhǎng)ist<T>是非線程安全集合，意思就是說所有的線程都可以修改他的值。

下面我們來看下并行集合 —— 線程安全集合，在System.Collections.Concurrent命名空間中，首先來看一下ConcurrentBag<T>泛型集合，其用法和List<T>類似，先來寫個(gè)方法測(cè)試一下：

public static void ConcurrentBagWithPallel()
  {
   ConcurrentBag<int> list = new ConcurrentBag<int>();
   Parallel.For(0, 10000, item =>
   {
   list.Add(item);
   });
   Console.WriteLine("ConcurrentBag's count is {0}", list.Count());
  }

同時(shí)執(zhí)行兩個(gè)方法，結(jié)果如下：

可以看到，ConcurrentBag集合的結(jié)果是正確的。下面我們修改代碼看看ConcurrentBag里面的數(shù)據(jù)到底是怎么存放的，修改代碼如下：

public static void ConcurrentBagWithPallel()
  {
   ConcurrentBag<int> list = new ConcurrentBag<int>();
   Parallel.For(0, 10000, item =>
   {
   list.Add(item);
   });
   Console.WriteLine("ConcurrentBag's count is {0}", list.Count());
   int n = 0;
   foreach(int i in list)
   {
   if (n > 10)
    break;
   n++;
   Console.WriteLine("Item[{0}] = {1}",n,i);
   }
   Console.WriteLine("ConcurrentBag's max item is {0}", list.Max());

  }

先來看一下運(yùn)行結(jié)果：

可以看到，ConcurrentBag中的數(shù)據(jù)并不是按照順序排列的，順序是亂的，隨機(jī)的。我們平時(shí)使用的Max、First、Last等linq方法都還有。其時(shí)分類似Enumerable的用法，大家可以參考微軟的MSDN了解它的具體用法。

關(guān)于線程安全的集合還有很多，和我們平時(shí)用的集合都差不多，比如類似Dictionary的ConcurrentDictionary，還有ConcurrentStack，ConcurrentQueue等。

二、Parallel Linq的用法及性能

1、AsParallel

前面了解了并行的For和foreach,今天就來看一下Linq的并行版本是怎么樣吧？為了測(cè)試，我們添加一個(gè)Custom類，代碼如下：

public class Custom
 {
  public string Name { get; set; }
  public int Age { get; set; }
  public string Address { get; set; }
 }

寫如下測(cè)試代碼：

public static void TestPLinq()
  {
   Stopwatch sw = new Stopwatch();
   List<Custom> customs = new List<Custom>();
   for (int i = 0; i < 2000000; i++)
   {
   customs.Add(new Custom() { Name = "Jack", Age = 21, Address = "NewYork" });
   customs.Add(new Custom() { Name = "Jime", Age = 26, Address = "China" });
   customs.Add(new Custom() { Name = "Tina", Age = 29, Address = "ShangHai" });
   customs.Add(new Custom() { Name = "Luo", Age = 30, Address = "Beijing" });
   customs.Add(new Custom() { Name = "Wang", Age = 60, Address = "Guangdong" });
   customs.Add(new Custom() { Name = "Feng", Age = 25, Address = "YunNan" });
   }

   sw.Start();
   var result = customs.Where<Custom>(c => c.Age > 26).ToList();
   sw.Stop();
   Console.WriteLine("Linq time is {0}.",sw.ElapsedMilliseconds);

   sw.Restart();
   sw.Start();
   var result2 = customs.AsParallel().Where<Custom>(c => c.Age > 26).ToList();
   sw.Stop();
   Console.WriteLine("Parallel Linq time is {0}.", sw.ElapsedMilliseconds);
  }

其實(shí)也就是加了一個(gè)AsParallel()方法，下面來看下運(yùn)行結(jié)果：

時(shí)間相差了一倍，不過有時(shí)候不會(huì)相差這么多，要看系統(tǒng)當(dāng)前的資源利用率。大家可以多測(cè)試一下。

其實(shí)，AsParallel()這個(gè)方法可以應(yīng)用與任何集合，包括List<T>集合，從而提高查詢速度和系統(tǒng)性能。

2、GroupBy方法

在項(xiàng)目中，我們經(jīng)常要對(duì)數(shù)據(jù)做處理，比如分組統(tǒng)計(jì)，我們知道在linq中也可以實(shí)現(xiàn)，今天來學(xué)習(xí)一下新的ToLookup方法，寫一個(gè)測(cè)試方法，代碼如下：

public static void OrderByTest()
  {
   Stopwatch stopWatch = new Stopwatch();
   List<Custom> customs = new List<Custom>();
   for (int i = 0; i < 2000000; i++)
   {
   customs.Add(new Custom() { Name = "Jack", Age = 21, Address = "NewYork" });
   customs.Add(new Custom() { Name = "Jime", Age = 26, Address = "China" });
   customs.Add(new Custom() { Name = "Tina", Age = 29, Address = "ShangHai" });
   customs.Add(new Custom() { Name = "Luo", Age = 30, Address = "Beijing" });
   customs.Add(new Custom() { Name = "Wang", Age = 60, Address = "Guangdong" });
   customs.Add(new Custom() { Name = "Feng", Age = 25, Address = "YunNan" });
   }

   stopWatch.Restart();
   var groupByAge = customs.GroupBy(item => item.Age).ToList();
   foreach (var item in groupByAge)
   {
   Console.WriteLine("Age={0},count = {1}", item.Key, item.Count());
   }
   stopWatch.Stop();

   Console.WriteLine("Linq group by time is: " + stopWatch.ElapsedMilliseconds);


   stopWatch.Restart();
   var lookupList = customs.ToLookup(i => i.Age);
   foreach (var item in lookupList)
   {
   Console.WriteLine("LookUP:Age={0},count = {1}", item.Key, item.Count());
   }
   stopWatch.Stop();
   Console.WriteLine("LookUp group by time is: " + stopWatch.ElapsedMilliseconds);
  }

運(yùn)行結(jié)果如下：