并行计算

Posted 2020-10-17 【我是谁】

一、Parallel.Invoke执行多个方法

没有特定执行顺序，利用cpu多核并发执行,在运行并行方法前都会产生一些额外的开销,我们一定要测量运行结果相比没有并行是否有优化

  #region Parallel.Invoke
            Stopwatch watch = new Stopwatch();
            watch.Start();
            Parallel.Invoke(Pinvoke1, Pinvoke2);
            watch.Stop();
            Console.WriteLine("Parallel run " + watch.ElapsedMilliseconds + " ms");
            watch.Reset();
            watch.Start();
            Pinvoke1();
            Pinvoke2();
            watch.Stop();
            Console.WriteLine("Normal run " + watch.ElapsedMilliseconds + " ms");
            Console.ReadKey();

            #endregion Parallel.Invoke

  private static void Pinvoke1()
        {
            Thread.Sleep(3000);
            Console.WriteLine("Pinvoke1 ThreadId="+Thread.CurrentContext.ContextID);
        }

        private static void Pinvoke2()
        {
            Thread.Sleep(2000);
            Console.WriteLine("Pinvoke2 ThreadId=" + Thread.CurrentContext.ContextID);
        }

　结果：

二、Parallel.For   每一次迭代一会当作一个任务交给一个线程，线程可能会重用，cpu决定

   Stopwatch watch = new Stopwatch();
            watch.Start();

            for (int i = 0; i < 5; i++)
            {
                Console.WriteLine(i);
                Thread.Sleep(1000);
            }
            watch.Stop();
            Console.WriteLine(watch.Elapsed);
       
            watch.Restart();
            Parallel.For(0, 5, i =>
            {
                Console.WriteLine("===" + Thread.CurrentThread.ManagedThreadId + "===");
                Console.WriteLine(i);
                Thread.Sleep(1000);
            });
            watch.Stop();
            Console.WriteLine(watch.Elapsed);
            Console.ReadKey();

效果还是非常明显的,但是有个问题是Parallel.For是并行执行,每次迭代基本都会启用新线程,那么我们共用一个全局变量的时候,必须要加锁,出现资源等待,这时用并行就不合适了:示例如下:

object asyncObj = new object();
            long value = 0;
            Stopwatch watch = new Stopwatch();
            watch.Start();
            for (int i = 0; i < 9999; i++)
            {
                for (int j = 0; j < 19999; j++)
                {
                    value++;
                }
            }
            watch.Stop();
            Console.WriteLine("正常运行是" + watch.ElapsedMilliseconds + " ms");
            watch.Restart();
            Parallel.For(0, 9999, i =>
            {
                for (int j = 0; j < 19999; j++)
                {
                    lock (asyncObj)
                    {
                        value++;
                    }
                }
            });
            watch.Stop();
            Console.WriteLine("并行运行是" + watch.ElapsedMilliseconds + " ms");
            Console.ReadKey();

三、Parallel.ForEach  都foreach一样，只不过是并行的，也是每次循环一个线程处理

    List<Person> list = new List<Person>();
            for (int i = 1; i < 100; i++)
            {
                list.Add(new Person
                {
                    Age = i,
                    Name = "Name" + i
                });
            }
            Stopwatch watch = new Stopwatch();
            watch.Start();
            Parallel.ForEach(list, p =>
            {
                Console.WriteLine("===" + Thread.CurrentThread.ManagedThreadId + "===");
                Console.WriteLine(p.Name);
            });
            watch.Stop();
            Console.ReadKey();

四、并行中断

ParallelLoopState:

1、Break-并行循环执行了当前迭代后尽快地停止执行。

//更改方法stop下面的示例,运行结果不一定哦,49,50都出现了

2、Stop-直接停止

//结果是50
  List<Person> list=new List<Person>();
            Parallel.For(0, 100, (num, state) =>
            {
                if (num<50)
                {
                    list.Add(new Person() {Age = num});
                }
                else
                {
                    state.Stop();
                }
            });
            Console.WriteLine(list.Count);

五、异常处理

并行执行的时候使用AggregateException获取全部异常,用Exception只能获取其中一个;

            try
            {
                Parallel.Invoke(Pinvoke1, Pinvoke2);
            }
            catch (AggregateException aggregateException)
            {
                foreach (var innerException in aggregateException.InnerExceptions)
                {
                    Console.WriteLine(innerException.Message);
                }
            }

    private static void Pinvoke1()
        {
            Thread.Sleep(3000);
            Console.WriteLine("Pinvoke1 ThreadId=" + Thread.CurrentContext.ContextID);
            throw new Exception("并行任务1抛出异常");
        }

        private static void Pinvoke2()
        {
            Thread.Sleep(2000);
            Console.WriteLine("Pinvoke2 ThreadId=" + Thread.CurrentContext.ContextID);
            throw new Exception("并行任务2抛出异常");
        }

 六、并行Linq （AsParallel，可以应用任何集合，提高处理速度）

  Stopwatch watch=new Stopwatch();
            List<Person> list = new List<Person>();
            Parallel.For(0,10000,p =>
            {
                list.Add(new Person() {Age = p,Name = "name"+p});//注意并行会出现有空对象的现象;也可以调用Invoke方法,里面写for循环就不会出现空对象
            });
            Console.WriteLine(list.Count);
            watch.Start();
            Console.WriteLine(list.Count);
            var persons = list.Where(c=>c!=null&&c.Age>100).ToList();
            watch.Stop();
            Console.WriteLine(watch.ElapsedMilliseconds);
            watch.Restart();
            var parallelPersons = list.AsParallel().Where(c=> c != null && c.Age>100).ToList();
            watch.Stop();
            Console.WriteLine(watch.ElapsedMilliseconds);
            Console.ReadKey();