看看Parallel中高度封装的三个方法,Invoke,For和ForEach

王朝学院·作者佚名  2016-08-27
窄屏简体版  字體: 小  |  中  |  大  |  超大  

说到.net中的并行编程,也许你的第一反应就是Task,确实Task是一个非常灵活的用于并行编程的一个专用类,不可否认越灵活的东西用起来就越

复杂,高度封装的东西用起来很简单,但是缺失了灵活性,这篇我们就看看这些好用但灵活性不高的几个并行方法。

一:Invoke

现在电子商务的网站都少不了订单的流程,没有订单的话网站也就没有存活的价值了,往往在订单提交成功后,通常会有这两个操作,第一个:发起

信用卡扣款,第二个:发送emial确认单,这两个操作我们就可以在下单接口调用成功后,因为两个方法是互不干扰的,所以就可以用invoke来玩玩了。

1staticvoidMain(string[] args)2{3Parallel.Invoke(Credit, Email);45Console.Read();6}78staticvoidCredit()9{10Console.WriteLine("****************** 发起信用卡扣款中 ******************");1112Thread.Sleep(2000);1314Console.WriteLine("扣款成功!");15}1617staticvoidEmail()18{19Console.WriteLine("****************** 发送邮件确认单!*****************");2021Thread.Sleep(3000);2223Console.WriteLine("email发送成功!");24}

怎么样,实现起来是不是很简单,只要把你需要的方法塞给invoke就行了,不过在这个方法里面有一个重载参数需要注意下,

1publicstaticvoidInvoke(ParallelOptions parallelOptions,paramsAction[] actions);

有时候我们的线程可能会跑遍所有的内核,为了提高其他应用程序的稳定性,就要限制参与的内核,正好ParallelOptions提供了

MaxDegreeOfParallelism属性。

好了,下面我们大概翻翻invoke里面的代码实现,发现有几个好玩的地方:

<1>: 当invoke中的方法超过10个话,我们发现它走了一个internal可见的ParallelForReplicatingTask的FCL内部专用类,而这个类是继承自

Task的,当方法少于10个的话,才会走常规的Task.

<2> 居然发现了一个装exception 的ConcurrentQueue<Exception>队列集合,多个异常入队后,再包装成AggregateException抛出来。

比如:throw new AggregateException(exceptionQ);

<3> 我们发现,不管是超过10个还是小于10个,都是通过WaitAll来等待所有的执行,所以缺点就在这个地方,如果某一个方法执行时间太长

不能退出,那么这个方法是不是会长期挂在这里不能出来,也就导致了主流程一直挂起,然后页面就一直挂起,所以这个是一个非常危险

的行为,如果我们用task中就可以在waitall中设置一个过期时间,但invoke却没法做到,所以在使用invoke的时候要慎重考虑。

1try2{3if(actionsCopy.Length>10|| (parallelOptions.MaxDegreeOfParallelism != -1&& parallelOptions.MaxDegreeOfParallelism <actionsCopy.Length))4{5ConcurrentQueue<Exception> exceptionQ = null;6try7{8intactionIndex =0;9ParallelForReplicatingTask parallelForReplicatingTask =newParallelForReplicatingTask(parallelOptions,delegate10{11for(intl = Interlocked.Increment(refactionIndex); l <= actionsCopy.Length; l = Interlocked.Increment(refactionIndex))12{13try14{15actionsCopy[l -1]();16}17catch(Exception item)18{19LazyInitializer.EnsureInitialized<ConcurrentQueue<Exception>>(refexceptionQ, () =>newConcurrentQueue<Exception>());20exceptionQ.Enqueue(item);21}22if(parallelOptions.CancellationToken.IsCancellationRequested)23{24thrownewOperationCanceledException(parallelOptions.CancellationToken);25}26}27}, TaskCreationOptions.None, InternalTaskOptions.SelfReplicating);28parallelForReplicatingTask.RunSynchronously(parallelOptions.EffectiveTaskScheduler);29parallelForReplicatingTask.Wait();30}31catch(Exception ex2)32{33LazyInitializer.EnsureInitialized<ConcurrentQueue<Exception>>(refexceptionQ, () =>newConcurrentQueue<Exception>());34AggregateException ex = ex2asAggregateException;35if(ex !=null)36{37using(IEnumerator<Exception> enumerator =ex.InnerExceptions.GetEnumerator())38{39while(enumerator.MoveNext())40{41Exception current =enumerator.Current;42exceptionQ.Enqueue(current);43}44gotoIL_264;45}46}47exceptionQ.Enqueue(ex2);48IL_264:;49}50if(exceptionQ !=null&& exceptionQ.Count >0)51{52Parallel.ThrowIfReducableToSingleOCE(exceptionQ, parallelOptions.CancellationToken);53thrownewAggregateException(exceptionQ);54}55}56else57{58Task[] array = new Task[actionsCopy.Length];59if(parallelOptions.CancellationToken.IsCancellationRequested)60{61thrownewOperationCanceledException(parallelOptions.CancellationToken);62}63for(intj =0; j < array.Length; j++)64{65array[j] =Task.Factory.StartNew(actionsCopy[j], parallelOptions.CancellationToken, TaskCreationOptions.None, InternalTaskOptions.None, parallelOptions.EffectiveTaskScheduler);66}67try68{69if(array.Length <=4)70{71Task.FastWaitAll(array);72}73else74{75Task.WaitAll(array);76}77}78catch(AggregateException ex3)79{80Parallel.ThrowIfReducableToSingleOCE(ex3.InnerExceptions, parallelOptions.CancellationToken);81throw;82}83finally84{85for(intk =0; k < array.Length; k++)86{87if(array[k].IsCompleted)88{89array[k].Dispose();90}91}92}93}94}95finally96{97if(TplEtwPRovider.Log.IsEnabled())98{99TplEtwProvider.Log.ParallelInvokeEnd((task !=null) ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, (task !=null) ? task.Id :0, forkJoinContextID);100}101}

二:For

下面再看看Parallel.For,我们知道普通的For是一个串行操作,如果说你的for中每条流程都需要执行一个方法,并且这些方法可以并行操作且

比较耗时,那么为何不尝试用Parallel.For呢,就比如下面的代码。

1classProgram2{3staticvoidMain(string[] args)4{5List<Action> actions =newList<Action>() { Credit, Email };67varresult = Parallel.For(0, actions.Count, (i) =>8{9actions[i]();10});1112Console.WriteLine("执行状态:"+result.IsCompleted);1314Console.Read();15}1617staticvoidCredit()18{19Console.WriteLine("****************** 发起信用卡扣款中 ******************");2021Thread.Sleep(2000);2223Console.WriteLine("扣款成功!");24}2526staticvoidEmail()27{28Console.WriteLine("****************** 发送邮件确认单!*****************");2930Thread.Sleep(3000);3132Console.WriteLine("email发送成功!");33}34}

下面我们再看看Parallel.For中的最简单的重载和最复杂的重载:

1publicstaticParallelLoopResult For(intfromInclusive,inttoExclusive, Action<int>body);23publicstaticParallelLoopResult For<TLocal>(intfromInclusive,inttoExclusive, ParallelOptions parallelOptions, Func<TLocal> localInit, Func<int, ParallelLoopState, TLocal, TLocal> body, Action<TLocal>localFinally);4

<1> 简单的重载不必多说,很简单,我上面的例子也演示了。

<2> 最复杂的这种重载提供了一个AOP的功能,在每一个body的action执行之前会先执行localInit这个action,在body之后还会执行localFinally

这个action,有没有感觉到已经把body切成了三块?好了,下面看一个例子。

1staticvoidMain(string[] args)2{3varlist =newList<int>() {10,20,30,40};45varoptions =newParallelOptions();67vartotal =0;89varresult = Parallel.For(0, list.Count, () =>10{11Console.WriteLine("------------ thead --------------");1213return1;14},15(i, loop, j) =>16{17Console.WriteLine("------------ body --------------");1819Console.WriteLine("i="+ list[i] +"j="+j);2021returnlist[i];22},23(i) =>24{25Console.WriteLine("------------ tfoot --------------");2627Interlocked.Add(reftotal, i);2829Console.WriteLine("total="+total);30});3132Console.WriteLine("iscompleted:"+result.IsCompleted);33Console.Read();34}

View Code

接下来我们再翻翻它的源代码,由于源码太多,里面神乎其神,我就找几个好玩的地方。

<1> 我在里面找到了一个rangeManager分区函数,代码复杂看不懂,貌似很强大。

1internalRangeManager(longnFromInclusive,longnToExclusive,longnStep,intnNumExpectedWorkers)2{3this.m_nCurrentIndexRangeToAssign =0;4this.m_nStep =nStep;5if(nNumExpectedWorkers ==1)6{7nNumExpectedWorkers =2;8}9ulongnum = (ulong)(nToExclusive -nFromInclusive);10ulongnum2 = num / (ulong)((long)nNumExpectedWorkers);11num2 -= num2 % (ulong)nStep;12if(num2 ==0uL)13{14num2 = (ulong)nStep;15}16intnum3 = (int)(num /num2);17if(num % num2 !=0uL)18{19num3++;20}21longnum4 = (long)num2;22this.m_indexRanges =newIndexRange[num3];23longnum5 =nFromInclusive;24for(inti =0; i < num3; i++)25{26this.m_indexRanges[i].m_nFromInclusive =num5;27this.m_indexRanges[i].m_nSharedCurrentIndexOffset =null;28this.m_indexRanges[i].m_bRangeFinished =0;29num5 +=num4;30if(num5 < num5 - num4 || num5 >nToExclusive)31{32num5 =nToExclusive;33}34this.m_indexRanges[i].m_nToExclusive =num5;35}36}

<2> 我又找到了这个神奇的ParallelForReplicatingTask类。

那么下面问题来了,在单线程的for中,我可以continue,可以break,那么在Parallel.For中有吗?因为是并行,所以continue基本上就没有

存在价值,break的话确实有价值,这个就是委托中的ParallelLoopState做到的,并且还新增了一个Stop。

三:ForEach

其实ForEach和for在本质上是一样的,你在源代码中会发现在底层都是调用一个方法的,而ForEach会在底层中调用for共同的函数之前还会执行

其他的一些逻辑,所以这就告诉我们,能用Parallel.For的地方就不要用Parallel.ForEach,其他的都一样了,这里就不赘述了。

 
 
免责声明:本文为网络用户发布,其观点仅代表作者个人观点,与本站无关,本站仅提供信息存储服务。文中陈述内容未经本站证实,其真实性、完整性、及时性本站不作任何保证或承诺,请读者仅作参考,并请自行核实相关内容。
 
© 2005- 王朝網路 版權所有 導航