前面看完了Task對(duì)象，這里再看一下另一個(gè)息息相關(guān)的對(duì)象Parallel。

Parallel對(duì)象

　　Parallel對(duì)象封裝了能夠利用多核并行執(zhí)行的多線程操作，其內(nèi)部使用Task來分裝多線程的任務(wù)并試圖將它們分配到不同的內(nèi)核中并行執(zhí)行。請(qǐng)注意“試圖”這個(gè)詞，Parallel對(duì)象相當(dāng)具有智能性，當(dāng)它判斷任務(wù)集并沒有從并行運(yùn)行中受益，就會(huì)選擇按順序運(yùn)行。這樣的做法是因?yàn)椴⒎撬械捻?xiàng)目都適合使用并行開發(fā)，創(chuàng)建過多并行任務(wù)可能會(huì)損害程序的性能，降低運(yùn)行效率。

　　Parallel對(duì)象是靜態(tài)類，它主要有3個(gè)靜態(tài)方法：Invoke，For，ForEach。針對(duì)這3個(gè)方法，該對(duì)象也提供了多種不同的重載方法，使用起來相當(dāng)?shù)暮唵巍Ｏ瓤匆粋€(gè)簡單的例子：

static void Main(string[] args) {Parallel.Invoke(()=>Console.WriteLine("1st task!"),()=>Console.WriteLine("2nd task!")); }

　　這個(gè)例子中的兩個(gè)任務(wù)就是并行執(zhí)行的，所以結(jié)果可能是第一個(gè)先完成，也可能是第二個(gè)先輸出結(jié)果。是不是超級(jí)簡單？有沒有使用一下Parallel對(duì)象的沖動(dòng)？

　　下面這個(gè)網(wǎng)上的例子驗(yàn)證了一下運(yùn)行時(shí)間上并行計(jì)算的優(yōu)越性：

private const int count = 1000000000; private static void M1() {Console.WriteLine("M1 is busy now");for (int i = 0; i < count; i++);Console.WriteLine("M1 is Done"); } private static void M2() {Console.WriteLine("M2 is busy now");for (int i = 0; i < count; i++);Console.WriteLine("M2 is Done"); } static void Main(string[] args) {// 順序執(zhí)行DateTime start1 = DateTime.Now;M1();M2();Console.WriteLine(DateTime.Now - start1);// 并行執(zhí)行DateTime start2 = DateTime.Now;Parallel.Invoke(M1, M2);Console.WriteLine(DateTime.Now - start2); }

　　在不同的機(jī)器上，得到的結(jié)果可能不同，但是基本上所有的多核機(jī)器上得到的結(jié)果一定是并行執(zhí)行的時(shí)候耗時(shí)比較短，例子比較簡單，但是道理確實(shí)很直接。

　　通常來說，對(duì)于一個(gè)程序，性能提升的關(guān)鍵是將可以并行執(zhí)行的同步程序改成并行執(zhí)行。這個(gè)上面的例子也反應(yīng)了修改后的效果。此外，對(duì)于程序來說，循環(huán)是影響復(fù)雜度的最直接的因素，這個(gè)我們看看教科書上計(jì)算算法時(shí)間復(fù)雜度的算法就知道了，所以提升循環(huán)的執(zhí)行效率往往是提升程序效率的關(guān)鍵一步。Parallel對(duì)象充分考慮到了這一點(diǎn)，提供了循環(huán)的并行版本。

例子一：For循環(huán)。

static void Main(string[] args) {for (int i = 0; i < 10; i++) Console.Write("{0} ", i);Console.WriteLine("by serial");Parallel.For(0, 10, (n) => Console.Write("{0} ", n));Console.WriteLine("by parallel"); }

　　從輸出的結(jié)果你可以很容易發(fā)現(xiàn)后面的結(jié)果順序完全是不固定的，這是并行的特征。

例子二：ForEach循環(huán)?

static void Main(string[] args) {int [] a = {1,2,3,4,5,6,7,8,9};foreach (var n in a) Console.Write("{0} ",n);Console.WriteLine("by serial");Parallel.ForEach(a, (n) => Console.Write("{0} ", n));Console.WriteLine("by parallel"); }

　　結(jié)果也很明顯，就不多說了。

　　通過上面的兩個(gè)例子，其實(shí)我們就能發(fā)現(xiàn)一些問題：

1. 順序要求嚴(yán)格的操作不能使用Parallel對(duì)象的方法，這個(gè)原因很簡單。

2.?并不是所有的for語句都可以用并行處理來實(shí)行，只有在循環(huán)開始前循環(huán)的次數(shù)已確定的情況下可以采用并行處理。同理，do語句和while語句也不能采用并行處理。因?yàn)樗^“并行”就是在判定為“循環(huán)結(jié)束”之前,首先要把將要執(zhí)行的循環(huán)實(shí)現(xiàn)分配好。

　　好了，既然是對(duì)循環(huán)的并行處理，那就避不開break與continue的問題，也就是循環(huán)的主動(dòng)中止問題。

循環(huán)的主動(dòng)中止

　　在Parallel對(duì)象中，也可以主動(dòng)中止循環(huán)的執(zhí)行：調(diào)用ParallelLoopState實(shí)例的Stop方法和Break方法，可以停止和中斷當(dāng)前循環(huán)的執(zhí)行。其中，

1. Break 告知 Parallel 循環(huán)應(yīng)在系統(tǒng)方便的時(shí)候盡早停止執(zhí)行當(dāng)前迭代之外的迭代，當(dāng)前迭代之前的迭代任然會(huì)完成。
2. Stop 告知 Parallel 循環(huán)應(yīng)在系統(tǒng)方便的時(shí)候盡早停止執(zhí)行，不管其他的線程執(zhí)行到什么程度。
　　通常使用Stop會(huì)立即停止循環(huán)，使用Break卻會(huì)執(zhí)行完畢當(dāng)前迭代次序前面的迭代后停止循環(huán)。例如，對(duì)于從 0 到 1000 并行迭代的 for 循環(huán)，如果從第 100 此迭代開始調(diào)用 Break，則低于 100 的所有迭代仍會(huì)運(yùn)行，從 101 到 1000 的迭代則不一定會(huì)執(zhí)行，注意是“不一定”，因?yàn)槭遣⑿袌?zhí)行的，說不定某些次序在后面的迭代已經(jīng)執(zhí)行了。看一下例子：

static void Main(string[] args) {DemoStop();DemoBreak(); }/// <summary> /// 中斷Stop /// </summary> static void DemoStop() {List<int> data = new List<int>(){ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9};Parallel.For(0, data.Count, (i, LoopState) =>{if (i > 5)LoopState.Stop();Thread.Sleep(500);Console.WriteLine(i);});Console.WriteLine("Stop執(zhí)行結(jié)束。"); } /// <summary> /// 中斷Break /// </summary> static void DemoBreak() {List<int> data = new List<int>() { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9};Parallel.ForEach(data, (i, LoopState) =>{if (i > 5)LoopState.Break();Thread.Sleep(500);Console.WriteLine(i);});Console.WriteLine("Break執(zhí)行結(jié)束。"); }

　　運(yùn)行一下，對(duì)比結(jié)果，細(xì)細(xì)體會(huì)一下輸出的結(jié)果，我想你就會(huì)清楚Stop方法與Break方法的區(qū)別。

　　當(dāng)然了，前面講的使用CancellationTokenSource取消線程的方式這里任然是適用的，不過需要通過ParallelOptions傳給Parallel對(duì)象對(duì)應(yīng)的重載方法。ParallelOptions對(duì)象還可以配置其他的一些參數(shù)，比如最大的并行數(shù)量(其實(shí)就是使用的最大內(nèi)核數(shù)量)等等。看一個(gè)簡單的例子：

CancellationTokenSource token = new CancellationTokenSource(); Task.Factory.StartNew(() => {Thread.Sleep(5000);token.Cancel();Console.WriteLine("Token Cancelled."); });ParallelOptions loopOptions = new ParallelOptions() {CancellationToken = token.Token,MaxDegreeOfParallelism = 2 };try {Parallel.For(0, Int64.MaxValue, loopOptions, i =>{Console.WriteLine("i={0},thread id={1}", i, Thread.CurrentThread.ManagedThreadId);Thread.Sleep(1000);}); } catch (OperationCanceledException) {Console.WriteLine("Exception..."); }

　　討論完了各種正常情況，下面來看一下不正常的情況：異常問題。

異常問題

　　和普通的for/foreach中發(fā)生異常的表現(xiàn)一樣，Parallel循環(huán)中的任何異常都會(huì)使整個(gè)循環(huán)終止，不過由于整個(gè)循環(huán)是分核同時(shí)進(jìn)行的，因此整個(gè)循環(huán)不會(huì)立即終止，這個(gè)很好理解。循環(huán)中停止前所有的異常都會(huì)被封裝在AggregateException的InnerExceptions中。捕獲這些異常的方式很簡單，使用try/catch就可以了，看一下下面的代碼：

try {Parallel.For(0, 5, (i) =>{throw new Exception(i.ToString());}); } catch (AggregateException ae) {foreach (var exp in ae.InnerExceptions){Console.WriteLine(exp.Message);} }

　　這段代碼將會(huì)輸出0-4的子集(也有可能是0-4全部輸出，因?yàn)?個(gè)線程都很快)。

　　不過，與Parallel.For和ForEach不一樣的是，Parallel.Invoke總是會(huì)把所有任務(wù)都執(zhí)行完，然后把所有的異常包裝在AggregateException中。其實(shí)道理與上面的循環(huán)是一樣的，都是把應(yīng)該執(zhí)行的任務(wù)執(zhí)行完，來看這段代碼：

try {Parallel.Invoke(() => { throw new Exception("1"); },() => { Thread.Sleep(1500); throw new Exception("2"); },() => { Thread.Sleep(3000); throw new Exception("3"); }); } catch (AggregateException ae) {foreach (var ex in ae.InnerExceptions){Console.WriteLine(ex.Message);} }

　　結(jié)果會(huì)輸出：3 2 1。

　　除此以外，Task.WaitAll和Parallel.Invoke是類似，任何一個(gè)(或多個(gè))Task的異常不會(huì)影響任何其他Task的執(zhí)行。

try {var t1 = Task.Factory.StartNew(() =>{Thread.Sleep(500);throw new Exception("1");});var t2 = Task.Factory.StartNew(() =>{Thread.Sleep(1000);throw new Exception("2");});Task.WaitAll(t1, t2); } catch (AggregateException ae) {foreach (var exp in ae.InnerExceptions){Console.WriteLine(exp.Message);} }

　　這段代碼會(huì)輸出：1 2。

　　兩個(gè)異常都會(huì)在AggregateException中的InnerExceptions屬性中。不過很顯然異常的順序與上一個(gè)例子有點(diǎn)不同，這個(gè)需要注意一點(diǎn)。

　　其實(shí)，在新的.NET類庫中，不僅通過增加Parallel對(duì)象來增強(qiáng)并行處理的能力，而且在Linq語句中也有相應(yīng)的增強(qiáng)，那就是PLinq。

PLinq簡介

　　PLINQ也就是Parallel Linq，它的使用方法是非常簡單。
　　下例本身沒有什么太大意義，只不過是找出“2”，然后輸出：

using System; using System.Linq; using System.Threading.Tasks; class Program {static void Main(string[] args){int[] ar = { 1, 2, 3 };var q1 = from n in arwhere n == 2select n;foreach (var n in q1){Console.WriteLine("found {0}", n);}} }

如果把上例改成用并行處理，只要在查詢表達(dá)式中追加AsParallel方法就可以了：

var q1 = from n in ar.AsParallel()where n == 2select n;

函數(shù)形式也是一樣的。例如下面這個(gè)查詢表達(dá)式：

var q1 = ar.Where((c) => c == 2);

改成并行執(zhí)行也就是插入AsParallel方法就可以了：

var q1 = ar.AsParallel().Where((c) => c == 2);

　　使用PLINQ是如此的簡單，只要用一個(gè)方法就可以用并行來處理查詢表達(dá)式了。但是，正如前面所講的并行計(jì)算并不是適用于任何場合的靈丹妙藥，它也有不太適用的場合：
1. 在大量使用查詢表達(dá)式的時(shí)候，并不是每一句查詢表達(dá)式都是性能瓶頸的關(guān)鍵，如果每一個(gè)查詢表達(dá)式都插入AsParallel方法，不會(huì)帶來太大好處，在浪費(fèi)時(shí)間的同時(shí)，代碼的可讀性也降低了。
2. 插入AsParallel方法后，結(jié)果會(huì)發(fā)生變化，這個(gè)自然很好理解，因?yàn)椴⑿袌?zhí)行了嘛，順序得不到保證，所以與順序有關(guān)的操作是適合使用同步操作的，并行執(zhí)行就可能導(dǎo)致問題。

　　其實(shí)AsParallel方法只是PLinq的基本入口點(diǎn)，在System.Linq.ParallelEnumerable類中，包含了并行查詢的大部分其他有用的方法，比如：AsSequential(指定查詢的其余部分應(yīng)像非并行 LINQ 查詢一樣按順序運(yùn)行)，AsOrdered(指定 PLINQ 應(yīng)保留查詢的其余部分的源序列排序，直到例如通過使用 orderby子句更改排序?yàn)橹?，AsUnordered(指定查詢的其余部分的 PLINQ 不需要保留源序列的排序)等等方法。這個(gè)查看一下MSDN就可以了，使用起來還是比較方便的。也可查看博客園中的一些詳細(xì)的文章，比如：http://www.cnblogs.com/leslies2/archive/2012/02/07/2320914.html。

?

　　并行計(jì)算就簡單總結(jié)這些了，銘記一點(diǎn)：并行執(zhí)行的任務(wù)要保證是順序無關(guān)的，獨(dú)立的。

總結(jié)

以上是生活随笔為你收集整理的C#的变迁史06 - C# 4.0 之并行处理篇的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。