C# 创建线程的多种方式之 Parallel类 基础知识

Parallel 类定义了并行运行的静态For(), Foreach(), Invoke()方法， 其中For(), Foreach() 多次调用同一个方法，方法返回值均为ParallelLoopResult，而Invoke()可同时调用多个不同的方法，无返回值。

For(), Foreach()方法有很多重载方法，以参数最多的For<TLocal>(Int32, Int32, ParallelOptions, Func<TLocal>, Func<Int32,ParallelLoopState,TLocal,TLocal>, Action<TLocal>) 进行说明，其他类似。以下参数含义说明来自MSDN，

类型参数

TLocal：The type of the thread-local data.     //局部变量，在

参数

fromInclusive Int32  ：The start index, inclusive.       //循环起始索引

toExclusive Int32 ：The end index, exclusive.           //循环结束索引

parallelOptions ParallelOptions ：An object that configures the behavior of this operation.        //并行执行参数，包括最大并发数量，线程取消操作通知，任务调度器

localInit Func<TLocal> ：The function delegate that returns the initial state of the local data for each task.     //每个线程初始化函数的委托

body Func<Int32,ParallelLoopState,TLocal,TLocal> ：The delegate that is invoked once per iteration.           //循环运行的函数体，ParallelLoopState  类可以调用Stop()和Break()方法，停止循环运行

localFinally Action<TLocal> ：The delegate that performs a final action on the local state of each task.          //每个线程结束处理函数的委托

为了方便观察，写了以下例子：

        static void Main(string[] args)
        {
            Console.WriteLine("Main Start....");
            ParallelOptions po = new ParallelOptions();      //并行执行参数类
            po.MaxDegreeOfParallelism = 4;        //最大并发数量
            int result=0;
            ParallelLoopResult pr = Parallel.For<int>(0, 20, po, () =>         //Initial
            {
                Console.WriteLine("taskId {0} Parallel initial.....",Task.CurrentId);     //输出当前任务Id
                return 1;
            },
            (i,ps,v1) =>                                                     //body
            {
                v1 += 1;
                Console.WriteLine("taskId {0} loopIndex{1},{2}", Task.CurrentId, i, v1);       //输出当前任务Id, 循环索引 , 局部变量
                return v1;
            },
            (v1) =>                                                             //finally
            {               
                Interlocked.Add(ref result, v1);                               //原子操作，操作共享变量result
                Console.WriteLine("taskId {0} Parallel finalized.....Result is {1}", Task.CurrentId, v1);    //输出当前任务Id, 局部变量
            });
            Console.WriteLine("Main end....{0}",result);
            Console.ReadLine();
        }

运行结果每次都有点不同，复制了一个运行结果，如下：

taskId 1 Parallel initial.....
taskId 1 loopIndex0,2
taskId 1 loopIndex1,3
taskId 2 Parallel initial.....
taskId 2 loopIndex5,2
taskId 2 loopIndex6,3
taskId 4 Parallel initial.....
taskId 4 loopIndex15,2
taskId 4 loopIndex16,3
taskId 4 loopIndex17,4
taskId 4 loopIndex18,5
taskId 4 loopIndex19,6
taskId 4 loopIndex3,7
taskId 4 loopIndex4,8
taskId 4 loopIndex8,9
taskId 4 loopIndex9,10
taskId 4 loopIndex11,11
taskId 4 loopIndex12,12
taskId 4 loopIndex13,13
taskId 4 loopIndex14,14
taskId 3 Parallel initial.....
taskId 1 loopIndex2,4
taskId 1 Parallel finalized.....Result is 4
taskId 3 loopIndex10,2
taskId 3 Parallel finalized.....Result is 2
taskId 2 loopIndex7,4
taskId 2 Parallel finalized.....Result is 4
taskId 4 Parallel finalized.....Result is 14
Main end....24

Paralle.For() 实际上是调用了线程池，至于调用线程的个数是不固定的，所以会导致每次运行显示不同。从上面的结果可以看出，此次运行共调用了4个线程，相应的 localInit 和 localFinally 也都调用了4次(注意: localInit 和 localFinally是指线程的初始化和结束处理，而不是循环)。从loopIndex的显示可以看出，循环执行是无顺序的。从局部变量v1的显示，可以看出局部变量在同一个线程中会由上次迭代传递给下次迭代，如上taskId 4 的局部变量一直在递增。

循环体中止：body方法参数 ParallelLoopState  类，调用stop() 或 break() 即可，但这两中方式有些不同。

Stop(): 停止当前迭代，阻止其他还未开始的迭代，已经在执行的迭代不受影响. 可以检测IsStop属性来判断是否调用Stop()方法，停止已经执行的迭代；   -- 推荐使用此方法停止

Break(): 阻止当前其他高于当前循环索引的迭代，已经在执行的不受影响。其实可以在每次迭代中检测ShouldExitCurrentIteration 属性，若为真，说明其他迭代调用了Break()，紧接着马上检测LowestBreakIteration 属性，如果当前循环索引小于该值，则立马return。

Parallel.Foreach() 的使用基本类似，ForEach<TSource,TLocal>(IEnumerable<TSource>, Func<TLocal>, Func<TSource,ParallelLoopState,TLocal,TLocal>, Action<TLocal>) ，

        static void Main(string[] args)
        {
            Console.WriteLine("Main Start....");
            List<int> numList = new List<int>();
            numList.AddRange(new int[]{0,1,2,3,4,5,6,7,8,9});
            int result=0;
            ParallelLoopResult pr = Parallel.ForEach<int, int>(numList, () =>         //Initial
            {
                Console.WriteLine("taskId {0} Parallel initial.....",Task.CurrentId);     //输出当前任务Id
                return 1;
            },
            (num,ps,i,v1) =>                                                     //body
            {
                v1 += num;
                Console.WriteLine("taskId {0} loopIndex{1},{2}", Task.CurrentId, i, v1);       //输出当前任务Id, 循环索引 , 局部变量
                return v1;
            },
            (v1) =>                                                             //finally
            {               
                Interlocked.Add(ref result, v1);                               //原子操作，操作共享变量result
                Console.WriteLine("taskId {0} Parallel finalized.....Result is {1}", Task.CurrentId, v1);    //输出当前任务Id, 局部变量
            });
            Console.WriteLine("Main end....{0}",result);
            Console.ReadLine();
        }

运行结果：

Main Start....
taskId 1 Parallel initial.....
taskId 2 Parallel initial.....
taskId 3 Parallel initial.....
taskId 4 Parallel initial.....
taskId 3 loopIndex4,5
taskId 3 loopIndex5,10
taskId 3 loopIndex7,17
taskId 3 loopIndex8,25
taskId 3 loopIndex9,34
taskId 3 loopIndex1,35
taskId 3 loopIndex3,38
taskId 3 Parallel finalized.....Result is 38
taskId 1 loopIndex0,1
taskId 1 Parallel finalized.....Result is 1
taskId 2 loopIndex2,3
taskId 2 Parallel finalized.....Result is 3
taskId 4 loopIndex6,7
taskId 4 Parallel finalized.....Result is 7
Main end....49

从结果分析基本上与For() 类似，但Foreach() 里还有一个 Partitioner 参数， 分区器。 通过分区器可以控制并发线程的数目，以及每个线程运行的迭代次数。

最后还有Invoke (params Action[] actions), 参数为params 可变参数，具体应用如下：

        static void Main(string[] args)
        {
            Console.WriteLine("Main Start....");
            Stopwatch sw = new Stopwatch();
            sw.Start();
            Parallel.Invoke(() =>
            {
                Console.WriteLine("TaskId{0} Func1 runs...",Task.CurrentId);
                Thread.Sleep(3000);
            },
            delegate
            {
                Console.WriteLine("TaskId{0} Func2 runs...", Task.CurrentId);
                Thread.Sleep(3000);
            },
            delegate
            {
                Console.WriteLine("TaskId{0} Func3 runs...", Task.CurrentId);
                Thread.Sleep(3000);
            });
            sw.Stop();
            Console.WriteLine("Main end....{0}",sw.Elapsed);
            Console.ReadLine();
        }

运行结果：

Main Start....
TaskId1 Func1 runs...
TaskId2 Func2 runs...
TaskId3 Func3 runs...
Main end....00:00:03.0503290

从结果看出，调用了3个不同的线程同时执行，虽然每个方法都延迟了3000ms，但总共运行时间也差不多是3000ms。