详细的.Net并行编程高级教程--Parallel

1、详细的.Net并行编程高级教程-Parallel一直觉得自己对并发了解不够深入，特别是看了代码整洁之道觉得自己有必要好好学学并发编程，因为性能也是衡量代码整洁的一大标准。而且在失控这本书中也多次提到并发，不管是计算机还是生物都并发处理着各种事物。人真是奇怪，当你关注一个事情的时候，你会发现周围的事物中就常出现那个事情。所以好奇心驱使下学习并发。便有了此文。一、理解硬件线程和软件线程   多核处理器带有一个以上的物理内核-物理内核是真正的独立处理单元，多个物理内核使得多条指令能够同时并行运行。硬件线程也称为逻辑内核，一个物理内 核可以使用超线程技术提供多个硬件线程。所以一个硬

2、件线程并不代表一个物理内核；Windows中每个运行的程序都是一个进程，每一个进程都会创建并运行 一个或多个线程，这些线程称为软件线程。硬件线程就像是一条泳道，而软件线程就是在其中游泳的人。二、并行场合  .Net Framework4 引入了新的Task Parallel Library(任务并行库，TPL)，它支持数据并行、任务并行和流水线。让开发人员应付不同的并行场合。· 数据并行：有大量数据需要处理，并且必须对每一份数据执行同样的操作。比如通过256bit的密钥对100个Unicode字符串进行AES算法加密。· 任务并行：通过任务并发运行不同的操作。例如

3、生成文件散列码，加密字符串，创建缩略图。· 流水线：这是任务并行和数据并行的结合体。  TPL引入了System.Threading.Tasks ,主类是Task，这个类表示一个异步的并发的操作，然而我们不一定要使用Task类的实例，可以使用Parallel静态类。它提供了 Parallel.Invoke, Parallel.For Parallel.Forecah 三个方法。三、Parallel.Invoke   试图让很多方法并行运行的最简单的方法就是使用Parallel类的Invoke方法。例如有四个方法：· WatchMovie

4、3; HaveDinner· ReadBook· WriteBlog  通过下面的代码就可以使用并行。System.Threading.Tasks.Parallel.Invoke(WatchMovie, HaveDinner, ReadBook, WriteBlog);这段代码会创建指向每一个方法的委托。Invoke方法接受一个Action的参数组。1public static void Invoke(params Action actions);用lambda表达式或匿名委托可以达到同样的效果。System.Threading.Tasks.Parallel.In

5、voke() => WatchMovie(), () => HaveDinner(), () => ReadBook(), delegate() WriteBlog(); ); 1.没有特定的执行顺序。 Parallel.Invoke方法只有在4个方法全部完成之后才会返回。它至少需要4个硬件线程才足以让这4个方法并发运行。但并不保证这4个方法能够同时启动运行，如果一个或者多个内核处于繁忙状态，那么底层的调度逻辑可能会延迟某些方法的初始化执行。给方法加上延时，就可以看到必须等待最长的方法执行完成才回到主方法。1. static void 

6、Main(string args) 2.          3.             System.Threading.Tasks.Parallel.Invoke(WatchMovie, HaveDinner, ReadBook, 4.         &

7、#160;       WriteBlog); 5.             Console.WriteLine("执行完成"); 6.             Console.ReadKey(); 7.    

8、;      8.  9.         static void WatchMovie() 10.          11.             Thread.Sleep(5000); 12.

9、            Console.WriteLine("看电影"); 13.          14.         static void HaveDinner() 15.      

10、60;   16.             Thread.Sleep(1000); 17.             Console.WriteLine("吃晚饭"); 18.          19.

11、         static void ReadBook() 20.          21.             Thread.Sleep(2000); 22.         &#

12、160;   Console.WriteLine("读书"); 23.          24.         static void WriteBlog() 25.          26.     

13、60;       Thread.Sleep(3000); 27.             Console.WriteLine("写博客"); 28.          这样会造成很多逻辑内核处于长时间闲置状态。四、Parallel.ForParallel.For为固定数目的独立

14、For循环迭代提供了负载均衡 (即将工作分发到不同的任务中执行，这样所有的任务在大部分时间都可以保持繁忙) 的并行执行。从而能尽可能地充分利用所有的可用的内核。我们比较下下面两个方法，一个使用For循环，一个使用Parallel.For  都是生成密钥在转换为十六进制字符串。1. private static void GenerateAESKeys() 2.          3.       

15、      var sw = Stopwatch.StartNew(); 4.             for (int i = 0; i < NUM_AES_KEYS; i+) 5.         &

16、#160;    6.                 var aesM = new AesManaged(); 7.                 aesM.GenerateKey();

17、 8.                 byte result = aesM.Key; 9.                 string hexStr = ConverToHexString(res

18、ult); 10.              11.             Console.WriteLine("AES:"+sw.Elapsed.ToString(); 12.          13.  

19、14.  private static void ParallelGenerateAESKeys() 15.          16.             var sw = Stopwatch.StartNew(); 17.       

20、;      System.Threading.Tasks.Parallel.For(1, NUM_AES_KEYS + 1, (int i) => 18.              19.             

21、;    var aesM = new AesManaged(); 20.                 aesM.GenerateKey(); 21.                &#

22、160;byte result = aesM.Key; 22.                 string hexStr = ConverToHexString(result); 23.             ); 24. &

23、#160;25.             Console.WriteLine("Parallel_AES:" + sw.Elapsed.ToString(); 26.          private static int NUM_AES_KEYS = 100000;      

24、  static void Main(string args)                    Console.WriteLine("执行"+NUM_AES_KEYS+"次："); GenerateAESKeys();            Parall

25、elGenerateAESKeys();            Console.ReadKey();        执行1000000次这里并行的时间是串行的一半。五、Parallel.ForEach在Parallel.For中，有时候对既有循环进行优化可能会是一个非常复杂的任务。Parallel.ForEach为固定数目的独立For Each循环迭代提供了负载均衡的并行执行，且支持自定义分区器，让使用者可以完全掌握

26、数据分发。实质就是将所有要处理的数据区分为多个部分，然后并行运 行这些串行循环。修改上面的代码：1. System.Threading.Tasks.Parallel.ForEach(Partitioner.Create(1, NUM_AES_KEYS + 1), range => 2.              3.        

27、         var aesM = new AesManaged(); 4.                 Console.WriteLine("AES Range(0,1 循环开始时间：2)",range.Item1,range.Item2,DateT

28、ime.Now.TimeOfDay); 5.  6.                 for (int i = range.Item1; i < range.Item2; i+) 7.            

29、60;     8.                     aesM.GenerateKey(); 9.                    &#

30、160;byte result = aesM.Key; 10.                     string hexStr = ConverToHexString(result); 11.           &

31、#160;      12.                 Console.WriteLine("AES:"+sw.Elapsed.ToString(); 13.             ); 从执行结果可以看出，分了

32、13个段执行的。 第二次执行还是13个段。速度上稍微有差异。开始没有指定分区数，Partitioner.Create使用的是内置默认值。而且我们发现这些分区并不是同时执行的，大致是分了三个时间段执行。而且执行顺序是不同的。总的时间和Parallel.For的方法差不多。public static ParallelLoopResult ForEach<TSource>(Partitioner<TSource> source, Action<TSource> body)Parallel.ForEach方法定义了source和Body两个参数。sourc

33、e是指分区器。提供了分解为多个分区的数据源。body是 要调用的委托。它接受每一个已定义的分区作为参数。一共有20多个重载，在上面的例子中，分区的类型为Tuple<int,int>,是一个 二元组类型。此外，返回一个ParallelLoopResult的值。Partitioner.Create 创建分区是根据逻辑内核数及其他因素决定。1. public static OrderablePartitioner<Tuple<int, int>> Create(int fromInclusive, int

34、 toExclusive) 2.      3.       int num = 3; 4.       if (toExclusive <= fromInclusive) 5.         throw new Argu

35、mentOutOfRangeException("toExclusive"); 6.       int rangeSize = (toExclusive - fromInclusive) / (PlatformHelper.ProcessorCount * num); 7.       if (rangeSize = 

36、;0) 8.         rangeSize = 1; 9.       return Partitioner.Create<Tuple<int, int>>(Partitioner.CreateRanges(fromInclusive, toExclusive, rangeSize), EnumerablePartitionerOptio

37、ns.NoBuffering); 10.      因此我们可以修改分区数目，rangesize大致为250000左右。也就是说我的逻辑内核是4.var rangesize = (int) (NUM_AES_KEYS/Environment.ProcessorCount) + 1;   System.Threading.Tasks.Parallel.ForEach(Partitioner.Create(1, NUM_AES_KEYS + 1,rangesize), range =>再次执行：分区变成了四个，时

38、间上没有多大差别（第一个时间是串行时间）。我们看见这四个分区几乎是同时执行的。大部分情况下，TPL在幕后使用的负载均衡机制都是非常高效的，然而对分区的控制便于使用者对自己的工作负载进行分析，来改进整体的性能。Parallel.ForEach也能对IEnumerable<int>集合进行重构。Enumerable.Range生产了序列化的数目。但这样就没有上面的分区效果。1. private static void ParallelForEachGenerateMD5HasHes() 2.     

39、60;    3.             var sw = Stopwatch.StartNew(); 4.             System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1, NUM_AES

40、_KEYS), number => 5.              6.                 var md5M = MD5.Create(); 7.       

41、          byte data = Encoding.Unicode.GetBytes(Environment.UserName + number); 8.                 byte result = md5M.Compute

42、Hash(data); 9.                 string hexString = ConverToHexString(result); 10.             ); 11.      

43、       Console.WriteLine("MD5:"+sw.Elapsed.ToString(); 12.          六、从循环中退出和串行运行中的break不同，ParallelLoopState 提供了两个方法用于停止Parallel.For 和 Parallel.ForEach的执行。· Break:让循环在执行了当前迭代后尽快停止执行。比如执行到100了，那么循环会处理掉

44、所有小于100的迭代。· Stop:让循环尽快停止执行。如果执行到了100的迭代，那不能保证处理完所有小于100的迭代。修改上面的方法：执行3秒后退出。1. private static void ParallelLoopResult(ParallelLoopResult loopResult) 2.          3.           &

45、#160; string text; 4.             if (loopResult.IsCompleted) 5.              6.            

46、60;    text = "循环完成" 7.              8.             else 9.            

47、;  10.                 if (loopResult.LowestBreakIteration.HasValue) 11.                  12.    

48、60;                text = "Break终止" 13.                  14.         &

49、#160;       else 15.                  16.                     text =&#

50、160;"Stop 终止" 17.                  18.              19.             Co

51、nsole.WriteLine(text); 20.          21.  22.  23.         private static void ParallelForEachGenerateMD5HasHesBreak() 24.          25.

52、            var sw = Stopwatch.StartNew(); 26.             var loopresult= System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1, NUM_AES_KE

53、YS), (int number,ParallelLoopState loopState) => 27.              28.                 var md5M = MD5.Create();&#

54、160;29.                 byte data = Encoding.Unicode.GetBytes(Environment.UserName + number); 30.                &#

55、160;byte result = md5M.ComputeHash(data); 31.                 string hexString = ConverToHexString(result); 32.            &#

56、160;    if (sw.Elapsed.Seconds > 3) 33.                  34.                   

57、60; loopState.Stop(); 35.                  36.             ); 37.             Pa

58、rallelLoopResult(loopresult); 38.             Console.WriteLine("MD5:" + sw.Elapsed); 39.          七、捕捉并行循环中发生的异常。当并行迭代中调用的委托抛出异常，这个异常没有在委托中被捕获到时，就会变成一组异常，新的System.A

59、ggregateException负责处理这一组异常。1. private static void ParallelForEachGenerateMD5HasHesException() 2.          3.             var sw = Stopwatch.StartNew(); 4.

60、            var loopresult = new ParallelLoopResult(); 5.             try 6.             &

61、#160;7.                 loopresult = System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1, NUM_AES_KEYS), (number, loopState) => 8.        &#

62、160;         9.                     var md5M = MD5.Create(); 10.            

63、0;        byte data = Encoding.Unicode.GetBytes(Environment.UserName + number); 11.                     byte result =

64、60;md5M.ComputeHash(data); 12.                     string hexString = ConverToHexString(result); 13.             &#

65、160;       if (sw.Elapsed.Seconds > 3) 14.                      15.            

66、60;            throw new TimeoutException("执行超过三秒"); 16.                      17.     

67、0;           ); 18.              19.             catch (AggregateException ex) 20.    

68、;          21.                 foreach (var innerEx in  ex.InnerExceptions) 22.           

69、       23.                     Console.WriteLine(innerEx.ToString(); 24.               

70、60;  25.              26.             27.             ParallelLoopResult(loopresult); 28.             Console.Wr