說到.net中的并行編程，也許你的第一反應(yīng)就是Task，確實Task是一個非常靈活的用于并行編程的一個專用類，不可否認(rèn)越靈活的東西用起來就越
復(fù)雜，高度封裝的東西用起來很簡單，但是缺失了靈活性，這篇我們就看看這些好用但靈活性不高的幾個并行方法。
一：Invoke
現(xiàn)在電子商務(wù)的網(wǎng)站都少不了訂單的流程，沒有訂單的話網(wǎng)站也就沒有存活的價值了，往往在訂單提交成功后，通常會有這兩個操作，第一個：發(fā)起
信用卡扣款，第二個：發(fā)送emial確認(rèn)單，這兩個操作我們就可以在下單接口調(diào)用成功后，因為兩個方法是互不干擾的，所以就可以用invoke來玩玩了。
1 static void Main(string[] args) 2 { 3 Parallel.Invoke(Credit, Email); 4 5 Console.Read(); 6 } 7 8 static void Credit() 9 {10 Console.WriteLine("****************** 發(fā)起信用卡扣款中 ******************");11 12 Thread.Sleep(2000);13 14 Console.WriteLine("扣款成功！");15 }16 17 static void Email()18 {19 Console.WriteLine("****************** 發(fā)送郵件確認(rèn)單！*****************");20 21 Thread.Sleep(3000);22 23 Console.WriteLine("email發(fā)送成功！");24 }
怎么樣，實現(xiàn)起來是不是很簡單，只要把你需要的方法塞給invoke就行了，不過在這個方法里面有一個重載參數(shù)需要注意下，
1 public static void Invoke(ParallelOptions parallelOptions, params Action[] actions);
有時候我們的線程可能會跑遍所有的內(nèi)核，為了提高其他應(yīng)用程序的穩(wěn)定性，就要限制參與的內(nèi)核，正好ParallelOptions提供了
MaxDegreeOfParallelism屬性。
好了，下面我們大概翻翻invoke里面的代碼實現(xiàn)，發(fā)現(xiàn)有幾個好玩的地方：
<1>: 當(dāng)invoke中的方法超過10個話，我們發(fā)現(xiàn)它走了一個internal可見的ParallelForReplicatingTask的FCL內(nèi)部專用類，而這個類是繼承自
Task的，當(dāng)方法少于10個的話，才會走常規(guī)的Task.
<2> 居然發(fā)現(xiàn)了一個裝exception 的ConcurrentQueue<Exception>隊列集合，多個異常入隊后，再包裝成AggregateException拋出來。
比如：throw new AggregateException(exceptionQ);
<3> 我們發(fā)現(xiàn)，不管是超過10個還是小于10個，都是通過WaitAll來等待所有的執(zhí)行，所以缺點就在這個地方，如果某一個方法執(zhí)行時間太長
不能退出，那么這個方法是不是會長期掛在這里不能出來，也就導(dǎo)致了主流程一直掛起，然后頁面就一直掛起，所以這個是一個非常危險
的行為，如果我們用task中就可以在waitall中設(shè)置一個過期時間，但invoke卻沒法做到，所以在使用invoke的時候要慎重考慮。
1 try 2 { 3 if (actionsCopy.Length > 10 || (parallelOptions.MaxDegreeOfParallelism != -1 && parallelOptions.MaxDegreeOfParallelism < actionsCopy.Length)) 4 { 5 ConcurrentQueue<Exception> exceptionQ = null; 6 try 7 { 8 int actionIndex = 0; 9 ParallelForReplicatingTask parallelForReplicatingTask = new ParallelForReplicatingTask(parallelOptions, delegate 10 { 11 for (int l = Interlocked.Increment(ref actionIndex); l <= actionsCopy.Length; l = Interlocked.Increment(ref actionIndex)) 12 { 13 try 14 { 15 actionsCopy[l - 1](); 16 } 17 catch (Exception item) 18 { 19 LazyInitializer.EnsureInitialized<ConcurrentQueue<Exception>>(ref exceptionQ, () => new ConcurrentQueue<Exception>()); 20 exceptionQ.Enqueue(item); 21 } 22 if (parallelOptions.CancellationToken.IsCancellationRequested) 23 { 24 throw new OperationCanceledException(parallelOptions.CancellationToken); 25 } 26 } 27 }, TaskCreationOptions.None, InternalTaskOptions.SelfReplicating); 28 parallelForReplicatingTask.RunSynchronously(parallelOptions.EffectiveTaskScheduler); 29 parallelForReplicatingTask.Wait(); 30 } 31 catch (Exception ex2) 32 { 33 LazyInitializer.EnsureInitialized<ConcurrentQueue<Exception>>(ref exceptionQ, () => new ConcurrentQueue<Exception>()); 34 AggregateException ex = ex2 as AggregateException; 35 if (ex != null) 36 { 37 using (IEnumerator<Exception> enumerator = ex.InnerExceptions.GetEnumerator()) 38 { 39 while (enumerator.MoveNext()) 40 { 41 Exception current = enumerator.Current; 42 exceptionQ.Enqueue(current); 43 } 44 goto IL_264; 45 } 46 } 47 exceptionQ.Enqueue(ex2); 48 IL_264:; 49 } 50 if (exceptionQ != null && exceptionQ.Count > 0) 51 { 52 Parallel.ThrowIfReducableToSingleOCE(exceptionQ, parallelOptions.CancellationToken); 53 throw new AggregateException(exceptionQ); 54 } 55 } 56 else 57 { 58 Task[] array = new Task[actionsCopy.Length]; 59 if (parallelOptions.CancellationToken.IsCancellationRequested) 60 { 61 throw new OperationCanceledException(parallelOptions.CancellationToken); 62 } 63 for (int j = 0; j < array.Length; j++) 64 { 65 array[j] = Task.Factory.StartNew(actionsCopy[j], parallelOptions.CancellationToken, TaskCreationOptions.None, InternalTaskOptions.None, parallelOptions.EffectiveTaskScheduler); 66 } 67 try 68 { 69 if (array.Length <= 4) 70 { 71 Task.FastWaitAll(array); 72 } 73 else 74 { 75 Task.WaitAll(array); 76 } 77 } 78 catch (AggregateException ex3) 79 { 80 Parallel.ThrowIfReducableToSingleOCE(ex3.InnerExceptions, parallelOptions.CancellationToken); 81 throw; 82 } 83 finally 84 { 85 for (int k = 0; k < array.Length; k++) 86 { 87 if (array[k].IsCompleted) 88 { 89 array[k].Dispose(); 90 } 91 } 92 } 93 } 94 } 95 finally 96 { 97 if (TplEtwPRovider.Log.IsEnabled()) 98 { 99 TplEtwProvider.Log.ParallelInvokeEnd((task != null) ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, (task != null) ? task.Id : 0, forkJoinContextID);100 }101 }
二：For
下面再看看Parallel.For，我們知道普通的For是一個串行操作，如果說你的for中每條流程都需要執(zhí)行一個方法，并且這些方法可以并行操作且
比較耗時，那么為何不嘗試用Parallel.For呢，就比如下面的代碼。
1 class Program 2 { 3 static void Main(string[] args) 4 { 5 List<Action> actions = new List<Action>() { Credit, Email }; 6 7 var result = Parallel.For(0, actions.Count, (i) => 8 { 9 actions[i]();10 });11 12 Console.WriteLine("執(zhí)行狀態(tài):" + result.IsCompleted);13 14 Console.Read();15 }16 17 static void Credit()18 {19 Console.WriteLine("****************** 發(fā)起信用卡扣款中 ******************");20 21 Thread.Sleep(2000);22 23 Console.WriteLine("扣款成功！");24 }25 26 static void Email()27 {28 Console.WriteLine("****************** 發(fā)送郵件確認(rèn)單！*****************");29 30 Thread.Sleep(3000);31 32 Console.WriteLine("email發(fā)送成功！");33 }34 }
下面我們再看看Parallel.For中的最簡單的重載和最復(fù)雜的重載：
1 public static ParallelLoopResult For(int fromInclusive, int toExclusive, Action<int> body);2 3 public static ParallelLoopResult For<TLocal>(int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Func<TLocal> localInit, Func<int, ParallelLoopState, TLocal, TLocal> body, Action<TLocal> localFinally);4
<1> 簡單的重載不必多說，很簡單，我上面的例子也演示了。
<2> 最復(fù)雜的這種重載提供了一個AOP的功能，在每一個body的action執(zhí)行之前會先執(zhí)行l(wèi)ocalInit這個action，在body之后還會執(zhí)行l(wèi)ocalFinally
這個action，有沒有感覺到已經(jīng)把body切成了三塊？好了，下面看一個例子。
1 static void Main(string[] args) 2 { 3 var list = new List<int>() { 10, 20, 30, 40 }; 4 5 var options = new ParallelOptions(); 6 7 var total = 0; 8 9 var result = Parallel.For(0, list.Count, () =>10 {11 Console.WriteLine("------------ thead --------------");12 13 return 1;14 },15 (i, loop, j) =>16 {17 Console.WriteLine("------------ body --------------");18 19 Console.WriteLine("i=" + list[i] + " j=" + j);20 21 return list[i];22 },23 (i) =>24 {25 Console.WriteLine("------------ tfoot --------------");26 27 Interlocked.Add(ref total, i);28 29 Console.WriteLine("total=" + total);30 });31 32 Console.WriteLine("iscompleted:" + result.IsCompleted);33 Console.Read();34 }
接下來我們再翻翻它的源代碼，由于源碼太多，里面神乎其神，我就找?guī)讉€好玩的地方。
<1> 我在里面找到了一個rangeManager分區(qū)函數(shù)，代碼復(fù)雜看不懂，貌似很強(qiáng)大。
1 internal RangeManager(long nFromInclusive, long nToExclusive, long nStep, int nNumExpectedWorkers) 2 { 3 this.m_nCurrentIndexRangeToAssign = 0; 4 this.m_nStep = nStep; 5 if (nNumExpectedWorkers == 1) 6 { 7 nNumExpectedWorkers = 2; 8 } 9 ulong num = (ulong)(nToExclusive - nFromInclusive);10 ulong num2 = num / (ulong)((long)nNumExpectedWorkers);11 num2 -= num2 % (ulong)nStep;12 if (num2 == 0uL)13 {14 num2 = (ulong)nStep;15 }16 int num3 = (int)(num / num2);17 if (num % num2 != 0uL)18 {19 num3++;20 }21 long num4 = (long)num2;22 this.m_indexRanges = new IndexRange[num3];23 long num5 = nFromInclusive;24 for (int i = 0; i < num3; i++)25 {26 this.m_indexRanges[i].m_nFromInclusive = num5;27 this.m_indexRanges[i].m_nSharedCurrentIndexOffset = null;28 this.m_indexRanges[i].m_bRangeFinished = 0;29 num5 += num4;30 if (num5 < num5 - num4 || num5 > nToExclusive)31 {32 num5 = nToExclusive;33 }34 this.m_indexRanges[i].m_nToExclusive = num5;35 }36 }
<2> 我又找到了這個神奇的ParallelForReplicatingTask類。
那么下面問題來了，在單線程的for中，我可以continue，可以break，那么在Parallel.For中有嗎？因為是并行，所以continue基本上就沒有
存在價值，break的話確實有價值，這個就是委托中的ParallelLoopState做到的，并且還新增了一個Stop。
三：ForEach
其實ForEach和for在本質(zhì)上是一樣的，你在源代碼中會發(fā)現(xiàn)在底層都是調(diào)用一個方法的，而ForEach會在底層中調(diào)用for共同的函數(shù)之前還會執(zhí)行
其他的一些邏輯，所以這就告訴我們，能用Parallel.For的地方就不要用Parallel.ForEach，其他的都一樣了，這里就不贅述了。
新聞熱點
疑難解答
