详细的.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 HaveDinner ReadBook WriteBlog 通过下面的代码就可以使用并行。System.Threading.T

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

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

6、teLine(执行完成);6. Console.ReadKey();7. 8. 9. staticvoidWatchMovie()10. 11. Thread.Sleep(5000);12. Console.WriteLine(看电影);13. 14. staticvoidHaveDinner()15. 16. Thread.Sleep(1000);17. Console.WriteLine(吃晚饭);18. 19. staticvoidReadBook()20. 21. Thread.Sleep(2000);22. Console.WriteLine(读书);23. 24. staticvo

7、idWriteBlog()25. 26. Thread.Sleep(3000);27. Console.WriteLine(写博客);28. 这样会造成很多逻辑内核处于长时间闲置状态。四、Parallel.ForParallel.For为固定数目的独立For循环迭代提供了负载均衡 (即将工作分发到不同的任务中执行，这样所有的任务在大部分时间都可以保持繁忙) 的并行执行。从而能尽可能地充分利用所有的可用的内核。我们比较下下面两个方法，一个使用For循环，一个使用Parallel.For 都是生成密钥在转换为十六进制字符串。1. privatestaticvoidGenerateAESKeys()

8、2. 3. varsw=Stopwatch.StartNew();4. for(inti=0;i18. 19. varaesM=newAesManaged();20. aesM.GenerateKey();21. byteresult=aesM.Key;22. stringhexStr=ConverToHexString(result);23. );24. 25. Console.WriteLine(Parallel_AES:+sw.Elapsed.ToString();26. private static int NUM_AES_KEYS = 100000; static void Main

9、(string args) Console.WriteLine(执行+NUM_AES_KEYS+次：); GenerateAESKeys(); ParallelGenerateAESKeys(); Console.ReadKey(); 执行1000000次这里并行的时间是串行的一半。五、Parallel.ForEach在Parallel.For中，有时候对既有循环进行优化可能会是一个非常复杂的任务。Parallel.ForEach为固定数目的独立For Each循环迭代提供了负载均衡的并行执行，且支持自定义分区器，让使用者可以完全掌握数据分发。实质就是将所有要处理的数据区分为多个部分，然后并行

10、运 行这些串行循环。修改上面的代码：1. System.Threading.Tasks.Parallel.ForEach(Partitioner.Create(1,NUM_AES_KEYS+1),range=2. 3. varaesM=newAesManaged();4. Console.WriteLine(AESRange(0,1循环开始时间：2),range.Item1,range.Item2,DateTime.Now.TimeOfDay);5. 6. for(inti=range.Item1;irange.Item2;i+)7. 8. aesM.GenerateKey();9. byte

11、result=aesM.Key;10. stringhexStr=ConverToHexString(result);11. 12. Console.WriteLine(AES:+sw.Elapsed.ToString();13. );从执行结果可以看出，分了13个段执行的。第二次执行还是13个段。速度上稍微有差异。开始没有指定分区数，Partitioner.Create使用的是内置默认值。而且我们发现这些分区并不是同时执行的，大致是分了三个时间段执行。而且执行顺序是不同的。总的时间和Parallel.For的方法差不多。public static ParallelLoopResult For

12、Each(Partitioner source, Action body)Parallel.ForEach方法定义了source和Body两个参数。source是指分区器。提供了分解为多个分区的数据源。body是 要调用的委托。它接受每一个已定义的分区作为参数。一共有20多个重载，在上面的例子中，分区的类型为Tuple,是一个 二元组类型。此外，返回一个ParallelLoopResult的值。Partitioner.Create 创建分区是根据逻辑内核数及其他因素决定。1. publicstaticOrderablePartitionerTupleCreate(intfromInclusiv

13、e,inttoExclusive)2. 3. intnum=3;4. if(toExclusive=fromInclusive)5. thrownewArgumentOutOfRangeException(toExclusive);6. intrangeSize=(toExclusive-fromInclusive)/(PlatformHelper.ProcessorCount*num);7. if(rangeSize=0)8. rangeSize=1;9. returnPartitioner.CreateTuple(Partitioner.CreateRanges(fromInclusive

14、,toExclusive,rangeSize),EnumerablePartitionerOptions.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 =再次执行：分区变成了四个，时间上

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

16、.ForEach(Enumerable.Range(1,NUM_AES_KEYS),number=5. 6. varmd5M=MD5.Create();7. bytedata=Encoding.Unicode.GetBytes(Environment.UserName+number);8. byteresult=md5M.ComputeHash(data);9. stringhexString=ConverToHexString(result);10. );11. Console.WriteLine(MD5:+sw.Elapsed.ToString();12. 六、从循环中退出和串行运行中的b

17、reak不同，ParallelLoopState 提供了两个方法用于停止Parallel.For 和 Parallel.ForEach的执行。 Break:让循环在执行了当前迭代后尽快停止执行。比如执行到100了，那么循环会处理掉所有小于100的迭代。 Stop:让循环尽快停止执行。如果执行到了100的迭代，那不能保证处理完所有小于100的迭代。修改上面的方法：执行3秒后退出。1. privatestaticvoidParallelLoopResult(ParallelLoopResultloopResult)2. 3. stringtext;4. if(loopResult.IsComple

18、ted)5. 6. text=循环完成;7. 8. else9. 10. if(loopResult.LowestBreakIteration.HasValue)11. 12. text=Break终止;13. 14. else15. 16. text=Stop终止;17. 18. 19. Console.WriteLine(text);20. 21. 22. 23. privatestaticvoidParallelForEachGenerateMD5HasHesBreak()24. 25. varsw=Stopwatch.StartNew();26. varloopresult=Syste

19、m.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1,NUM_AES_KEYS),(intnumber,ParallelLoopStateloopState)=27. 28. varmd5M=MD5.Create();29. bytedata=Encoding.Unicode.GetBytes(Environment.UserName+number);30. byteresult=md5M.ComputeHash(data);31. stringhexString=ConverToHexString(result);32. if(sw.El

20、apsed.Seconds3)33. 34. loopState.Stop();35. 36. );37. ParallelLoopResult(loopresult);38. Console.WriteLine(MD5:+sw.Elapsed);39. 七、捕捉并行循环中发生的异常。当并行迭代中调用的委托抛出异常，这个异常没有在委托中被捕获到时，就会变成一组异常，新的System.AggregateException负责处理这一组异常。1. privatestaticvoidParallelForEachGenerateMD5HasHesException()2. 3. varsw=Stop

21、watch.StartNew();4. varloopresult=newParallelLoopResult();5. try6. 7. loopresult=System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1,NUM_AES_KEYS),(number,loopState)=8. 9. varmd5M=MD5.Create();10. bytedata=Encoding.Unicode.GetBytes(Environment.UserName+number);11. byteresult=md5M.ComputeHash(

22、data);12. stringhexString=ConverToHexString(result);13. if(sw.Elapsed.Seconds3)14. 15. thrownewTimeoutException(执行超过三秒);16. 17. );18. 19. catch(AggregateExceptionex)20. 21. foreach(varinnerExinex.InnerExceptions)22. 23. Console.WriteLine(innerEx.ToString();24. 25. 26. 27. ParallelLoopResult(loopresult);28. Console.WriteLine(MD5:+sw.Elapsed);29. 结果：异常出现了好几次。八、指定并行度。TPL的方法总会试图利用所有可用的逻辑内核来实现最好的结果，但有时候你并不希望在并行循环中使用所有的内核。比如你需要留出一个不参与并行计算 的内核，来创建能够响应用户的应用程序，而且这个内核需要帮助你运行代码中的其他部分。这个时候一种好的解决方法就是指定最大并行度。这需要创建一个ParallelOptions的实例，设置M