C 40 新特性之并行运算(Parallel)

1、.NET(C#) Internals: 以一个数组填充的例子初步了解.NET 4.0中的并行（一）引言随着CPU多核的普及，编程时充分利用这个特性越显重要。本文首先用传统的嵌套循环进行数组填充，然后用.NET 4.0中的System.Threading.Tasks提供的Parallel Class来并行地进行填充（当然这里也用到嵌套循环），通过对比发现其中差异。主要内容如下：· 通常的数组填充 · 并行的组数填充 · 性能比较 · System.Threading.Tasks分析，这个将在续篇.NET(C#) Internals: 以一个数组填充的例子初

2、步了解.NET 4.0中的并行（二）中介绍1、通常的数组填充首先看如下代码：通常的数组填充using System;namespace ParallelForSample public class SingleCore public static void Calculate(int calcVal) Utility util = new Utility(); util.Start(); int, G = new intcalcVal, calcVal; for (int k = 0; k < calcVal; k+) for (int i = 0; i < calcVal; i+

3、) for (int j = 0; j < calcVal; j+) Gi, j = Math.Min(Gi, j, Gi, k + Gk, j); util.Stop(); 上面的粗体红色显示的几行代码就是实现数组填充，这个很好理解不用多费口舌。补充说明的是：上面的Utility是为了统计性能而编写的一个类，它主要就是用到了Stopwatch对象它提供一组方法和属性，可用于准确地测量运行时间。Utility的代码如下：Utility类 public class Utility private Stopwatch _stopwatch; public void Start() _stop

4、watch = new Stopwatch(); _stopwatch.Start(); public void Stop() _stopwatch.Stop(); TimeSpan ts = _stopwatch.Elapsed; string elapsedTime = String.Format("0:00:1:00:2:00.3:00", ts.Hours, ts.Minutes, ts.Seconds, ts.Milliseconds / 10); Console.WriteLine("Time taken : 0", elapsedTime)

5、; 利用它我们就可以对数组填充所耗费的时间进行计算了。 2、并行的组数填充为了充分利用CPU的多核，我们编写如下代码：并行的数组填充using System;using System.Threading.Tasks;namespace ParallelForSample public class MultiCore public static void Calculate(int calcVal) Utility util = new Utility(); util.Start(); int, G = new intcalcVal, calcVal; Parallel.For(0, calcV

6、al, delegate(int k) Parallel.For(0, calcVal, delegate(int i) for (int j = 0; j < calcVal; j+) Gi, j = Math.Min(Gi, j, Gi, k + Gk, j); ); ); util.Stop(); 留意上面的红色粗体显示的几行代码，它利用了Parallel.For Method (Int32, Int32, Action<Int32>)方法，Parallel类位于命名空间System.Threading.Tasks中，它支持并行循环。此Parallel.For方法使得它

7、里面的迭代可能并行地运行，注意到上述代码中它的第三个参数是一个委托。在（0，calcVal）之间，这个委托将被调用。3、性能比较现在我们来测试一下，上面两种方法的执行性能差异如何，下载源码。其实，核心代码已经在上面贴出来了，现在注意是编写实例来测试，代码主要如下： 性能比较测试using System;using System.Collections.Generic;using System.Linq;using System.Text;namespace ParallelForSample class Program static void Main(string args) Console

8、.WriteLine("Single core"); SingleCore.Calculate(1000); Console.WriteLine("Multi core"); MultiCore.Calculate(1000); Console.WriteLine("Finished"); Console.ReadKey(); 运行之后得到如下结果：（不同电脑配置不同，得出结果不同）图1、性能比较从结果可以看出，并行的数组填充比通常的数组填充性能更高。  System.Threading.Tasks分析，这个将在续篇.NET

9、(C#) Internals: 以一个数组填充的例子初步了解.NET 4.0中的并行（二）中介绍.NET(C#) Internals: 以一个数组填充的例子初步了解.NET 4.0中的并行（二）引言随着CPU多核的普及，编程时充分利用这个特性越显重要。上篇首先用传统的嵌套循环进行数组填充，然后用.NET 4.0中的System.Threading.Tasks提供的Parallel Class来并行地进行填充，最后对比他们的性能。本文将深入分析Parallel Class并借机回答上篇9楼提出的问题，而System.Threading.Tasks分析，这个将推迟到.NET(C#) Interna

10、ls: 以一个数组填充的例子初步了解.NET 4.0中的并行（三）中介绍。内容如下：· 1、Parallel Class o 1.1、For方法 o 1.2、ForEach方法 o 1.3、Invoke方法 · 2、并发控制疑问？ o 2.1、使用Lock锁o 2.2、使用PLINQ用AsParallelo 2.3、使用PLINQ用ParallelEnumerable o 2.4、使用Interlocked操作 o 2.5、使用Parallel.For的有Thread-Local变量重载函数· 性能比较1、Parallel ClassParallel这个类提供对通

11、常操作（诸如for、foreach、执行语句块）基于库的数据并行替换。它只是System.Threading.Tasks命名空间的一个类，该命名空间中还包括很多其他的类。下面举个例子来说明如何使用Parallel.For（来自MSDN）： 01using System.Threading.Tasks;    02class Test 03 04    static int N = 1000; 05  06    static void TestMethod() 07

12、     08        / Using a named method. 09        Parallel.For(0, N, Method2); 10  11        / Using an anonymous method. 12    

13、60;   Parallel.For(0, N, delegate(int i) 13         14            / Do Work. 15        ); 16  17      

14、60; / Using a lambda expression. 18        Parallel.For(0, N, i => 19         20            / Do Work. 21        

15、;); 22     23  24    static void Method2(int i) 25     26        / Do work. 27     28上面这个例子简单易懂，上篇我们就是用的Parallel.For,这里就不解释了。其实Parallel类的方法主要分为下面三类：· For方法 · Fo

16、rEach方法 · Invoke方法 1.1、For方法在里面执行的for循环可能并行地运行，它有12个重载。这12个重载中Int32参数和Int64参数的方法各为6个，下面以Int32为例列出：· For(Int32 fromInclusive, Int32 toExclusive, Action<Int32> body)，该方法对区间（fromInclusive，toExclusive）之间的迭代调用body表示的委托。body委托有一个迭代数次的int32参数，如果fromInclusive>=toExclusive，则不会执行任何迭代。 ·

17、; For(Int32 fromInclusive, Int32 toExclusive, Action<Int32, ParallelLoopState>)，该方法对区间(fromInclusive, toExclusive)之间的迭代调用body表示的委托。body委托有两个参数表示迭代数次的int32参数、一个可用于过早地跳出循环的ParallelLoopState实例。如果fromInclusive>=toExclusive，则不会执行任何迭代。 调用Break通知For操作当前迭代之后的迭代不需要执行。然而，在此之前的迭代如果没有完成仍然需要执行。因此，调用Brea

18、k类似于调用break跳出传统的for循环，不是break的原因是它不保证当前迭代之后的迭代绝对不会执行。 如果在当前迭代之前的迭代不必要执行，应该调用Stop而不是Break。调用Stop通知For循环放弃剩下的迭代，不管它是否在当前迭代之前或之后，因为所以要求的工作已经完成。然而，Break并不能保证这个。 · For(Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Action<Int32> body)，跟第一个方法类似，但它的区间是fromInclusive, toE

19、xclusive)。 · For(Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Action<Int32, ParallelLoopState> body)，跟第二个方法类似，单的区间是fromInclusive, toExclusive)。 · For<TLocal>(Int32 fromInclusive, Int32 toExclusive, Func<TLocal> localInit, Func<Int32, Paralle

20、lLoopState, TLocal, TLocal> body, Action<TLocal> localFinally)，它的迭代区间是fromInclusive, toExclusive)。另外body有两个local状态变量用于同一线程的迭代之间共享。localInit委托将在每个线程参与循环执行时调用，并返回这些线程初始的local状态。这些初始状态被传递给body，当它在每个线程上第一次调用时。然后，接下来body调用返回一个可能的修改状态值且传递给下一次body调用。最终，最后一次在每个线程上的body调用返回的一个状态值传递给localFinally委托。每个

21、线程执行在自己的loacl 状态上执行最后一个动作时，localFinally委托将被调用。这个委托可能在多个线程上并发执行，因此，你必须同步访问任何共享变量。 · For<TLocal>(Int32, Int32, ParallelOptions, Func<TLocal>, Func<Int32, ParallelLoopState, TLocal, TLocal>, Action<TLocal>)，跟上面的方法类似。 下面代码演示了For(Int32 fromInclusive, Int32 toExclusive, Parall

22、elOptions parallelOptions, Action<Int32> body)方法（来自MSDN）：01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplication2 09 10    class Prog

23、ram 11     12        / Demonstrated features: 13        /        CancellationTokenSource 14        /    &

24、#160;    Parallel.For() 15        /        ParallelOptions 16        /        ParallelLoopResult 17      

25、  / Expected results: 18        /         An iteration for each argument value (0, 1, 2, 3, 4, 5, 6, 7, 8, 9) is executed. 19        /      

26、  The order of execution of the iterations is undefined. 20        /        The iteration when i=2 cancels the loop. 21        /        Some iter

27、ations may bail out or not start at all; because they are temporally executed in unpredictable order,  22        /          it is impossible to say which will start/complete and which won't. 23  &

28、#160;     /        At the end, an OperationCancelledException is surfaced. 24        / Documentation: 25        /        26&#

29、160; 27        static void Main(string args) 28         29            CancellationTokenSource cancellationSource = new CancellationTokenSource(); 30 &

30、#160;          ParallelOptions options = new ParallelOptions(); 31            options.CancellationToken = cancellationSource.Token; 32         &#

31、160;  try33             34                ParallelLoopResult loopResult = Parallel.For( 35          

32、          0, 36                    10, 37                  &

33、#160; options, 38                    (i, loopState) => 39                     40  

34、                      Console.WriteLine("Start Thread=0, i=1", Thread.CurrentThread.ManagedThreadId, i); 41  42          

35、;              / Simulate a cancellation of the loop when i=2 43                        if (i = 2) 44 

36、60;                       45                           

37、 cancellationSource.Cancel(); 46                         47  48                 

38、       / Simulates a long execution 49                        for (int j = 0; j < 10; j+) 50        &#

39、160;                51                            Thread.Sleep(1 * 200); 52 

40、0;53                            / check to see whether or not to continue 54              

41、              if (loopState.ShouldExitCurrentIteration) return; 55                         56  57

42、60;                       Console.WriteLine("Finish Thread=0, i=1", Thread.CurrentThread.ManagedThreadId, i); 58          

43、60;          59                ); 60                if (loopResult.IsCompleted) 61  &

44、#160;              62                    Console.WriteLine("All iterations completed successfully. THIS WAS NOT EXPECTED.");

45、63                 64             65                / No exception is expected i

46、n this example, but if one is still thrown from a task, 66                / it will be wrapped in AggregateException and propagated to the main thread. 67           

47、 catch (AggregateException e) 68             69                Console.WriteLine("Parallel.For has thrown an AggregateException. THIS WAS NOT EXPECTED

48、.n0", e); 70             71                / Catching the cancellation exception 72            

49、catch (OperationCanceledException e) 73             74                Console.WriteLine("An iteration has triggered a cancellation. THIS WAS EXPECTED.n0&qu

50、ot;, e.ToString(); 75             76         77     781.2、ForEach方法在迭代中执行的foreach操作可能并行地执行，它有20个重载。这个方法太多，但用法大概跟For方法差不多，请自行参考MSDN。1.3、Invoke方法提供的每个动作可能并行地执行，它有2个重载。· Invoke(p

51、arams Action actions)：actions是一个要执行的动作数组，这些动作可能并行地执行，但并不保证执行的顺序及一定并行执行。这个方法直到提供的所有操作完成时才返回，不管是否正常地完成或异常终止。 · Invoke(ParallelOptions parallelOptions, params Action actions)：跟上面的方法类似，只是增加了一个parallelOptions参数，可以用户调用者取消整个操作。 例如下面代码执行了三个操作（来自MSDN）： 01using System; 02using System.Collections.Gen

52、eric; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplication2 09 10    class Program 11     12        static void Main() 13 &

53、#160;       14            try15             16                Parallel

54、.Invoke( 17                    BasicAction,    / Param #0 - static method 18                  &

55、#160; () =>            / Param #1 - lambda expression 19                     20         

56、0;              Console.WriteLine("Method=beta, Thread=0", Thread.CurrentThread.ManagedThreadId); 21                    

57、, 22                    delegate()        / Param #2 - in-line delegate 23               &

58、#160;     24                        Console.WriteLine("Method=gamma, Thread=0", Thread.CurrentThread.ManagedThreadId); 25     

59、;                26                ); 27             28     

60、       / No exception is expected in this example, but if one is still thrown from a task, 29            / it will be wrapped in AggregateException and propagated to the main thread. 30   &#

61、160;        catch (AggregateException e) 31             32                Console.WriteLine("An action has thrown a

62、n exception. THIS WAS UNEXPECTED.n0", e.InnerException.ToString(); 33             34         35  36        static void BasicAction() 37 

63、0;       38            Console.WriteLine("Method=alpha, Thread=0", Thread.CurrentThread.ManagedThreadId); 39         40     412、并发控制疑问？有人提出以

64、下疑问：“如果For里面的东西，对于顺序敏感的话，会不会有问题。并行处理的话，说到底应该是多线程。如果需要Lock住什么东西的话，应该怎么做呢？例如这个例子不是对数组填充，是对文件操作呢？对某个资源操作呢？”关于对顺序敏感的话，也就是说该如何加锁来控制？下面我举个例子来说明：对11000求和。如果我们想上篇那样简单地用Parallel.For，将会产生错误的结果，代码如下：01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Thr

65、eading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplication2 09 10    class Program 11             12        static void Main(string args) 13  &#

66、160;      14            int loops=0; 15            while (loops <= 100) 16             17

67、60;               long sum = 0;                 18                

68、;Parallel.For(1, 1001, delegate(long i) 19                 20                    sum += i; 21    

69、60;           ); 22                System.Console.WriteLine(sum); 23                l

70、oops+; 24             25         26     27在上述代码中，为了校验正确性我进行了重复做了100次，得出如下结果：图1、100次的前面部分结果我们知道500500才是正确的答案，这说明Parallel.For不能保证对sum正确的并发执行，对此我们应该加上适当的控制，并借机来回答上面提出的如何加锁的问题。下面有几种方案可

71、以解决这个问题：2.1、使用Lock锁 这个我就不多解释了，直接上代码：01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplication2 09 10    class Program 11   

72、60; 12        static void Main(string args) 13         14            int loops = 0; 15            ob

73、ject moniter = new object(); 16            while (loops <= 100) 17             18                

74、long sum = 0; 19                Parallel.For(1, 1001, delegate(long i) 20                 21        

75、            lock (moniter) sum += i; 22                ); 23                Syst

76、em.Console.WriteLine(sum); 24                loops+; 25             26         27     28我们加上lock锁之后就

77、会得出正确的结果。2.2、使用PLINQ用AsParallel关于PLINQ，以后将会介绍到，这里不会详细介绍，感兴趣的自行查阅资料。代码如下： 01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplication2 09 10   &

78、#160;class Program 11     12        static void Main(string args) 13         14            int loops = 0; 15     &#

79、160;      while (loops <= 100) 16             17                long sum = 0;      18   

80、;             sum = Enumerable.Range(0, 1001).AsParallel().Sum(); 19                System.Console.WriteLine(sum); 20      &#

81、160;         loops+; 21             22         23     24运行可以得到正确的结果。2.3、使用PLINQ用ParallelEnumerable 这个也不多说，直接上代码，因为关于PLINQ将在以后详细介绍，感兴趣的自

82、行查阅资料。01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplication2 09 10    class Program 11     12    

83、    static void Main(string args) 13         14            int loops = 0; 15            while (loops <= 100) 16 

84、            17                long sum = 0; 18                sum = ParallelEnumerabl

85、e.Range(0, 1001).Sum();  19                System.Console.WriteLine(sum); 20                loops+; 21     

86、60;       22         23     24运行同样可以得到正确结果。2.4、使用Interlocked操作 代码如下： 01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.T

87、hreading.Tasks; 07  08namespace ConsoleApplication2 09 10    class Program 11     12        static void Main(string args) 13         14      

88、;      int loops = 0; 15            while (loops <= 100) 16             17            &

89、#160;   long sum = 0; 18                Parallel.For(1, 1001, delegate(long i) 19                 20    &

90、#160;               Interlocked.Add(ref sum, i); 21                ); 22            &

91、#160;   System.Console.WriteLine(sum); 23                loops+; 24             25  26        

92、27     28运行可以得到正确结果。2.5、使用Parallel.For的有Thread-Local变量重载函数 这个方法已经在1.2中介绍，这里直接上代码，代码如下：01using System; 02using System.Collections.Generic; 03using System.Linq; 04using System.Text; 05using System.Threading; 06using System.Threading.Tasks; 07  08namespace ConsoleApplicati

93、on2 09 10    class Program 11     12        static void Main(string args) 13         14            int loops = 0; 15

94、0;           while (loops <= 100) 16             17                int sum = 0; 18   &#

95、160;            Parallel.For(0, 1001, () => 0, (i, state,subtotal) => 19                 20          

96、60;         subtotal += i; 21                    return subtotal; 22              

97、60; , 23                partial => Interlocked.Add(ref sum, partial); 24  25                System.Console.WriteLine(sum); 26