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

一直觉得自己对并发了解不够深入，特别是看了《代码整洁之道》觉得自己有必要好好学学并发编程，因为性能也是衡量代码整洁的一大标准。而且在《失控》这本书中也多次提到并发，不管是计算机还是生物都并发处理着各种事物。人真是奇怪，当你关注一个事情的时候，你会发现周围的事物中就常出现那个事情。所以好奇心驱使下学习并发。便有了此文。

一、理解硬件线程和软件线程

多核处理器带有一个以上的物理内核--物理内核是真正的独立处理单元，多个物理内核使得多条指令能够同时并行运行。硬件线程也称为逻辑内核，一个物理内核可以使用超线程技术提供多个硬件线程。所以一个硬件线程并不代表一个物理内核；Windows中每个运行的程序都是一个进程，每一个进程都会创建并运行一个或多个线程，这些线程称为软件线程。硬件线程就像是一条泳道，而软件线程就是在其中游泳的人。

二、并行场合

.Net Framework4 引入了新的Task Parallel Library(任务并行库，TPL)，它支持数据并行、任务并行和流水线。让开发人员应付不同的并行场合。

数据并行：有大量数据需要处理，并且必须对每一份数据执行同样的操作。比如通过256bit的密钥对100个Unicode字符串进行AES算法加密。
任务并行：通过任务并发运行不同的操作。例如生成文件散列码，加密字符串，创建缩略图。
流水线：这是任务并行和数据并行的结合体。

TPL引入了System.Threading.Tasks ,主类是Task，这个类表示一个异步的并发的操作，然而我们不一定要使用Task类的实例，可以使用Parallel静态类。它提供了 Parallel.Invoke, Parallel.For Parallel.Forecah 三个方法。

三、Parallel.Invoke

试图让很多方法并行运行的最简单的方法就是使用Parallel类的Invoke方法。例如有四个方法：

WatchMovie
HaveDinner
ReadBook
WriteBlog

通过下面的代码就可以使用并行。

System.Threading.Tasks.Parallel.Invoke(WatchMovie, HaveDinner, ReadBook, WriteBlog);

这段代码会创建指向每一个方法的委托。Invoke方法接受一个Action的参数组。

1	`public` `static` `void` `Invoke(params` `Action[] actions);`

用lambda表达式或匿名委托可以达到同样的效果。

System.Threading.Tasks.Parallel.Invoke(() => WatchMovie(), () => HaveDinner(), () => ReadBook(), delegate() { WriteBlog(); });

1.没有特定的执行顺序。

Parallel.Invoke方法只有在4个方法全部完成之后才会返回。它至少需要4个硬件线程才足以让这4个方法并发运行。但并不保证这4个方法能够同时启动运行，如果一个或者多个内核处于繁忙状态，那么底层的调度逻辑可能会延迟某些方法的初始化执行。

给方法加上延时，就可以看到必须等待最长的方法执行完成才回到主方法。

 
 
  
  static void Main(string[] args)
  
          {
  
              System.Threading.Tasks.Parallel.Invoke(WatchMovie, HaveDinner, ReadBook,
  
                  WriteBlog);
  
              Console.WriteLine("执行完成");
  
              Console.ReadKey();
  
          }
  
  
  
          static void WatchMovie()
  
          {
  
              Thread.Sleep(5000);
  
              Console.WriteLine("看电影");
  
          }
  
          static void HaveDinner()
  
          {
  
              Thread.Sleep(1000);
  
              Console.WriteLine("吃晚饭");
  
          }
  
          static void ReadBook()
  
          {
  
              Thread.Sleep(2000);
  
              Console.WriteLine("读书");
  
          }
  
          static void WriteBlog()
  
          {
  
              Thread.Sleep(3000);
  
              Console.WriteLine("写博客");
  
          }

这样会造成很多逻辑内核处于长时间闲置状态。

四、Parallel.For

Parallel.For为固定数目的独立For循环迭代提供了负载均衡 (即将工作分发到不同的任务中执行，这样所有的任务在大部分时间都可以保持繁忙) 的并行执行。从而能尽可能地充分利用所有的可用的内核。

我们比较下下面两个方法，一个使用For循环，一个使用Parallel.For 都是生成密钥在转换为十六进制字符串。

  
 
   
  private static void GenerateAESKeys()
   
          {
   
              var sw = Stopwatch.StartNew();
   
              for (int i = 0; i < NUM_AES_KEYS; i++)
   
              {
   
                  var aesM = new AesManaged();
   
                  aesM.GenerateKey();
   
                  byte[] result = aesM.Key;
   
                  string hexStr = ConverToHexString(result);
   
              }
   
              Console.WriteLine("AES:"+sw.Elapsed.ToString());
   
          }
   
  
   
   private static void ParallelGenerateAESKeys()
   
          {
   
              var sw = Stopwatch.StartNew();
   
              System.Threading.Tasks.Parallel.For(1, NUM_AES_KEYS + 1, (int i) =>
   
              {
   
                  var aesM = new AesManaged();
   
                  aesM.GenerateKey();
   
                  byte[] result = aesM.Key;
   
                  string hexStr = ConverToHexString(result);
   
              });
   
  
   
              Console.WriteLine("Parallel_AES:" + sw.Elapsed.ToString());
   
          }

private static int NUM_AES_KEYS = 100000;
        static void Main(string[] args)
        {
            Console.WriteLine("执行"+NUM_AES_KEYS+"次："); GenerateAESKeys();
            ParallelGenerateAESKeys();
            Console.ReadKey();
        }

执行1000000次

这里并行的时间是串行的一半。

五、Parallel.ForEach

在Parallel.For中，有时候对既有循环进行优化可能会是一个非常复杂的任务。Parallel.ForEach为固定数目的独立For Each循环迭代提供了负载均衡的并行执行，且支持自定义分区器，让使用者可以完全掌握数据分发。实质就是将所有要处理的数据区分为多个部分，然后并行运行这些串行循环。

修改上面的代码：

 
 
  
  System.Threading.Tasks.Parallel.ForEach(Partitioner.Create(1, NUM_AES_KEYS + 1), range =>
  
              {
  
                  var aesM = new AesManaged();
  
                  Console.WriteLine("AES Range({0},{1} 循环开始时间：{2})",range.Item1,range.Item2,DateTime.Now.TimeOfDay);
  
  
  
                  for (int i = range.Item1; i < range.Item2; i++)
  
                  {
  
                      aesM.GenerateKey();
  
                      byte[] result = aesM.Key;
  
                      string hexStr = ConverToHexString(result);
  
                  }
  
                  Console.WriteLine("AES:"+sw.Elapsed.ToString());
  
              });

从执行结果可以看出，分了13个段执行的。

第二次执行还是13个段。速度上稍微有差异。开始没有指定分区数，Partitioner.Create使用的是内置默认值。

而且我们发现这些分区并不是同时执行的，大致是分了三个时间段执行。而且执行顺序是不同的。总的时间和Parallel.For的方法差不多。

public static ParallelLoopResult ForEach(Partitioner source, Action body)

Parallel.ForEach方法定义了source和Body两个参数。source是指分区器。提供了分解为多个分区的数据源。body是要调用的委托。它接受每一个已定义的分区作为参数。一共有20多个重载，在上面的例子中，分区的类型为Tuple,是一个二元组类型。此外，返回一个ParallelLoopResult的值。

Partitioner.Create 创建分区是根据逻辑内核数及其他因素决定。

 
 
  
  public static OrderablePartitioner> Create(int fromInclusive, int toExclusive)
  
      {
  
        int num = 3;
  
        if (toExclusive <= fromInclusive)
  
          throw new ArgumentOutOfRangeException("toExclusive");
  
        int rangeSize = (toExclusive - fromInclusive) / (PlatformHelper.ProcessorCount * num);
  
        if (rangeSize == 0)
  
          rangeSize = 1;
  
        return Partitioner.Create>(Partitioner.CreateRanges(fromInclusive, toExclusive, rangeSize), EnumerablePartitionerOptions.NoBuffering);
  
      }

因此我们可以修改分区数目，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 =>

再次执行：

分区变成了四个，时间上没有多大差别（***个时间是串行时间）。我们看见这四个分区几乎是同时执行的。大部分情况下，TPL在幕后使用的负载均衡机制都是非常高效的，然而对分区的控制便于使用者对自己的工作负载进行分析，来改进整体的性能。

Parallel.ForEach也能对IEnumerable集合进行重构。Enumerable.Range生产了序列化的数目。但这样就没有上面的分区效果。

 
 
  
  private static void ParallelForEachGenerateMD5HasHes()
  
          {
  
              var sw = Stopwatch.StartNew();
  
              System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1, NUM_AES_KEYS), number =>
  
              {
  
                  var md5M = MD5.Create();
  
                  byte[] data = Encoding.Unicode.GetBytes(Environment.UserName + number);
  
                  byte[] result = md5M.ComputeHash(data);
  
                  string hexString = ConverToHexString(result);
  
              });
  
              Console.WriteLine("MD5:"+sw.Elapsed.ToString());
  
          }

#p#

六、从循环中退出

和串行运行中的break不同，ParallelLoopState 提供了两个方法用于停止Parallel.For 和 Parallel.ForEach的执行。

Break:让循环在执行了当前迭代后尽快停止执行。比如执行到100了，那么循环会处理掉所有小于100的迭代。
Stop:让循环尽快停止执行。如果执行到了100的迭代，那不能保证处理完所有小于100的迭代。

修改上面的方法：执行3秒后退出。

 
 
  
  private static void ParallelLoopResult(ParallelLoopResult loopResult)
  
          {
  
              string text;
  
              if (loopResult.IsCompleted)
  
              {
  
                  text = "循环完成";
  
              }
  
              else
  
              {
  
                  if (loopResult.LowestBreakIteration.HasValue)
  
                  {
  
                      text = "Break终止";
  
                  }
  
                  else
  
                  {
  
                      text = "Stop 终止";
  
                  }
  
              }
  
              Console.WriteLine(text);
  
          }
  
  
  
  
  
          private static void ParallelForEachGenerateMD5HasHesBreak()
  
          {
  
              var sw = Stopwatch.StartNew();
  
              var loopresult= System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1, NUM_AES_KEYS), (int number,ParallelLoopState loopState) =>
  
              {
  
                  var md5M = MD5.Create();
  
                  byte[] data = Encoding.Unicode.GetBytes(Environment.UserName + number);
  
                  byte[] result = md5M.ComputeHash(data);
  
                  string hexString = ConverToHexString(result);
  
                  if (sw.Elapsed.Seconds > 3)
  
                  {
  
                      loopState.Stop();
  
                  }
  
              });
  
              ParallelLoopResult(loopresult);
  
              Console.WriteLine("MD5:" + sw.Elapsed);
  
          }

七、捕捉并行循环中发生的异常。

当并行迭代中调用的委托抛出异常，这个异常没有在委托中被捕获到时，就会变成一组异常，新的System.AggregateException负责处理这一组异常。

 
 
  
  private static void ParallelForEachGenerateMD5HasHesException()
  
          {
  
              var sw = Stopwatch.StartNew();
  
              var loopresult = new ParallelLoopResult();
  
              try
  
              {
  
                  loopresult = System.Threading.Tasks.Parallel.ForEach(Enumerable.Range(1, NUM_AES_KEYS), (number, loopState) =>
  
                  {
  
                      var md5M = MD5.Create();
  
                      byte[] data = Encoding.Unicode.GetBytes(Environment.UserName + number);
  
                      byte[] result = md5M.ComputeHash(data);
  
                      string hexString = ConverToHexString(result);
  
                      if (sw.Elapsed.Seconds > 3)
  
                      {
  
                          throw new TimeoutException("执行超过三秒");
  
                      }
  
                  });
  
              }
  
              catch (AggregateException ex)
  
              {
  
                  foreach (var innerEx in  ex.InnerExceptions)
  
                  {
  
                      Console.WriteLine(innerEx.ToString());
  
                  }
  
              }
  
             
  
              ParallelLoopResult(loopresult);
  
              Console.WriteLine("MD5:" + sw.Elapsed);
  
          }

结果：

异常出现了好几次。

#p#

八、指定并行度。

TPL的方法总会试图利用所有可用的逻辑内核来实现***的结果，但有时候你并不希望在并行循环中使用所有的内核。比如你需要留出一个不参与并行计算的内核，来创建能够响应用户的应用程序，而且这个内核需要帮助你运行代码中的其他部分。这个时候一种好的解决方法就是指定***并行度。

这需要创建一个ParallelOptions的实例，设置MaxDegreeOfParallelism的值。

 
 
  
  private static void ParallelMaxDegree(int maxDegree)
  
          {
  
              var parallelOptions = new ParallelOptions();
  
              parallelOptions.MaxDegreeOfParallelism = maxDegree;
  
  
  
              var sw = Stopwatch.StartNew();
  
              System.Threading.Tasks.Parallel.For(1, NUM_AES_KEYS + 1, parallelOptions, (int i) =>
  
              {
  
                  var aesM = new AesManaged();
  
                  aesM.GenerateKey();
  
                  byte[] result = aesM.Key;
  
                  string hexStr = ConverToHexString(result);
  
              });
  
              Console.WriteLine("AES:" + sw.Elapsed.ToString());
  
          }

调用：如果在四核微处理器上运行，那么将使用3个内核。

ParallelMaxDegree(Environment.ProcessorCount - 1);

时间上大致慢了点（***次Parallel.For 3.18s），但可以腾出一个内核来处理其他的事情。

小结：这次学习了Parallel相关方法以及如何退出并行循环和捕获异常、设置并行度，还有并行相关的知识。园子里也有类似的博客。但作为自己知识的管理，在这里梳理一遍。

本文名称：详细的.Net并行编程高级教程--Parallel
文章URL：http://www.mswzjz.cn/qtweb/news4/423104.html

攀枝花网站建设、攀枝花网站运维推广公司-贝锐智能，是专注品牌与效果的网络营销公司；服务项目有等

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：贝锐智能

贝锐智能技术为您推荐以下文章