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

简介:

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

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

 

一:Invoke

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

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

static void Main(string[] args)
        {
            Parallel.Invoke(Credit, Email);

            Console.Read();
        }

        static void Credit()
        {
            Console.WriteLine("******************  发起信用卡扣款中  ******************");

            Thread.Sleep(2000);

            Console.WriteLine("扣款成功!");
        }

        static void Email()
        {
            Console.WriteLine("******************  发送邮件确认单!*****************");

            Thread.Sleep(3000);

            Console.WriteLine("email发送成功!");
        }

 

 

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

1  public static void Invoke(ParallelOptions parallelOptions, params Action[] 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的时候要慎重考虑。

try
    {
        if (actionsCopy.Length > 10 || (parallelOptions.MaxDegreeOfParallelism != -1 && parallelOptions.MaxDegreeOfParallelism < actionsCopy.Length))
        {
            ConcurrentQueue<Exception> exceptionQ = null;
            try
            {
                int actionIndex = 0;
                ParallelForReplicatingTask parallelForReplicatingTask = new ParallelForReplicatingTask(parallelOptions, delegate
                {
                    for (int l = Interlocked.Increment(ref actionIndex); l <= actionsCopy.Length; l = Interlocked.Increment(ref actionIndex))
                    {
                        try
                        {
                            actionsCopy[l - 1]();
                        }
                        catch (Exception item)
                        {
                            LazyInitializer.EnsureInitialized<ConcurrentQueue<Exception>>(ref exceptionQ, () => new ConcurrentQueue<Exception>());
                            exceptionQ.Enqueue(item);
                        }
                        if (parallelOptions.CancellationToken.IsCancellationRequested)
                        {
                            throw new OperationCanceledException(parallelOptions.CancellationToken);
                        }
                    }
                }, TaskCreationOptions.None, InternalTaskOptions.SelfReplicating);
                parallelForReplicatingTask.RunSynchronously(parallelOptions.EffectiveTaskScheduler);
                parallelForReplicatingTask.Wait();
            }
            catch (Exception ex2)
            {
                LazyInitializer.EnsureInitialized<ConcurrentQueue<Exception>>(ref exceptionQ, () => new ConcurrentQueue<Exception>());
                AggregateException ex = ex2 as AggregateException;
                if (ex != null)
                {
                    using (IEnumerator<Exception> enumerator = ex.InnerExceptions.GetEnumerator())
                    {
                        while (enumerator.MoveNext())
                        {
                            Exception current = enumerator.Current;
                            exceptionQ.Enqueue(current);
                        }
                        goto IL_264;
                    }
                }
                exceptionQ.Enqueue(ex2);
                IL_264:;
            }
            if (exceptionQ != null && exceptionQ.Count > 0)
            {
                Parallel.ThrowIfReducableToSingleOCE(exceptionQ, parallelOptions.CancellationToken);
                throw new AggregateException(exceptionQ);
            }
        }
        else
        {
            Task[] array = new Task[actionsCopy.Length];
            if (parallelOptions.CancellationToken.IsCancellationRequested)
            {
                throw new OperationCanceledException(parallelOptions.CancellationToken);
            }
            for (int j = 0; j < array.Length; j++)
            {
                array[j] = Task.Factory.StartNew(actionsCopy[j], parallelOptions.CancellationToken, TaskCreationOptions.None, InternalTaskOptions.None, parallelOptions.EffectiveTaskScheduler);
            }
            try
            {
                if (array.Length <= 4)
                {
                    Task.FastWaitAll(array);
                }
                else
                {
                    Task.WaitAll(array);
                }
            }
            catch (AggregateException ex3)
            {
                Parallel.ThrowIfReducableToSingleOCE(ex3.InnerExceptions, parallelOptions.CancellationToken);
                throw;
            }
            finally
            {
                for (int k = 0; k < array.Length; k++)
                {
                    if (array[k].IsCompleted)
                    {
                        array[k].Dispose();
                    }
                }
            }
        }
    }
    finally
    {
        if (TplEtwProvider.Log.IsEnabled())
        {
            TplEtwProvider.Log.ParallelInvokeEnd((task != null) ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, (task != null) ? task.Id : 0, forkJoinContextID);
        }
    }

 

二:For

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

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

class Program
    {
        static void Main(string[] args)
        {
            List<Action> actions = new List<Action>() { Credit, Email };

            var result = Parallel.For(0, actions.Count, (i) =>
            {
                actions[i]();
            });

            Console.WriteLine("执行状态:" + result.IsCompleted);

            Console.Read();
        }

        static void Credit()
        {
            Console.WriteLine("******************  发起信用卡扣款中  ******************");

            Thread.Sleep(2000);

            Console.WriteLine("扣款成功!");
        }

        static void Email()
        {
            Console.WriteLine("******************  发送邮件确认单!*****************");

            Thread.Sleep(3000);

            Console.WriteLine("email发送成功!");
        }
    }


下面我们再看看Parallel.For中的最简单的重载和最复杂的重载:
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> 最复杂的这种重载提供了一个AOP的功能,在每一个body的action执行之前会先执行localInit这个action,在body之后还会执行localFinally

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


static void Main(string[] args)
        {
            var list = new List<int>() { 10, 20, 30, 40 };

            var options = new ParallelOptions();

            var total = 0;

            var result = Parallel.For(0, list.Count, () =>
            {
                Console.WriteLine("------------  thead --------------");

                return 1;
            },
              (i, loop, j) =>
              {
                  Console.WriteLine("------------  body --------------");

                  Console.WriteLine("i=" + list[i] + " j=" + j);

                  return list[i];
              },
              (i) =>
              {
                  Console.WriteLine("------------  tfoot --------------");

                  Interlocked.Add(ref total, i);

                  Console.WriteLine("total=" + total);
              });

            Console.WriteLine("iscompleted:" + result.IsCompleted);
            Console.Read();
        }

 

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

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

internal RangeManager(long nFromInclusive, long nToExclusive, long nStep, int nNumExpectedWorkers)
        {
            this.m_nCurrentIndexRangeToAssign = 0;
            this.m_nStep = nStep;
            if (nNumExpectedWorkers == 1)
            {
                nNumExpectedWorkers = 2;
            }
            ulong num = (ulong)(nToExclusive - nFromInclusive);
            ulong num2 = num / (ulong)((long)nNumExpectedWorkers);
            num2 -= num2 % (ulong)nStep;
            if (num2 == 0uL)
            {
                num2 = (ulong)nStep;
            }
            int num3 = (int)(num / num2);
            if (num % num2 != 0uL)
            {
                num3++;
            }
            long num4 = (long)num2;
            this.m_indexRanges = new IndexRange[num3];
            long num5 = nFromInclusive;
            for (int i = 0; i < num3; i++)
            {
                this.m_indexRanges[i].m_nFromInclusive = num5;
                this.m_indexRanges[i].m_nSharedCurrentIndexOffset = null;
                this.m_indexRanges[i].m_bRangeFinished = 0;
                num5 += num4;
                if (num5 < num5 - num4 || num5 > nToExclusive)
                {
                    num5 = nToExclusive;
                }
                this.m_indexRanges[i].m_nToExclusive = num5;
            }
        }

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

 

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

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

 

 

三:ForEach

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

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

相关文章
|
2月前
|
存储 编译器 C语言
【C++ 模板编程 实用手段】深入理解 C++ 中的 packaged_task、invoke_result_t、bind、result_of 和 Lambda
【C++ 模板编程 实用手段】深入理解 C++ 中的 packaged_task、invoke_result_t、bind、result_of 和 Lambda
41 0
|
4月前
|
XML SQL JSON
query 与 params:选择正确的参数传递方式
query 与 params:选择正确的参数传递方式
|
9月前
手动封装call
手动封装call
32 0
|
9月前
|
前端开发
手动封装call
手动封装call
35 0
|
11月前
|
Java
JAVA中函数式接口-------Function ,BiFunction计算转换接口
JAVA中函数式接口-------Function ,BiFunction计算转换接口
|
Dart
Dart之 对象call方法
Dart之 对象call方法
158 0
Dart之 对象call方法
|
IDE 开发工具 Python
如何调用别的.air脚本中封装好的方法
如何调用别的.air脚本中封装好的方法
295 0
利用Java Lambda Function实现的自定义排序逻辑(Comparator)
利用Java Lambda Function实现的自定义排序逻辑(Comparator)
179 0
|
JavaScript
JS/NPAP之间传递数组:NPVARIANT_TO_OBJECT/OBJECT_TO_NPVARIANT
JS/NPAP之间传递数组:NPVARIANT_TO_OBJECT/OBJECT_TO_NPVARIANT
123 0