并行的扩展
扩展1.
Parallel的使用:
在Parallel下面有三个常用的方法Invoke,For,ForEach
Parallel.Invoke()方法是最简单,最简洁的将串行的代码并行化.
在这里先说一点,就是Stopwatch的使用,Stopwatch到底是个什么东西,首先Stopwatch在命名空间System.Diagnostics中.
使用方法如下:
var StopWatch =new Stopwatch();//创建一个Stopwatch实例
StopWatch.Start();//开始计时
StopWatch.Stop();//停止计时
StopWatch.Reset();//重置StopWatch
StopWatch.Restart();//重启被停止的Stopwatch
stopWatch.ElapsedMilliseconds //获取stopWatch从开始到现在的时间差，单位是毫秒
案例:
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace 并行集合和LINQ
{
class Program
{
static void Main(string[] args)
{
Custom custom = new Custom();
custom.ParallelInvokeMethod();
Console.WriteLine(“运行完成”);
Console.ReadKey();
}
}
public class Custom
{
private Stopwatch stopWatch = new Stopwatch();
public void Run1()
{
Thread.Sleep(2000);//Thread需要添加命名空间
Console.WriteLine(“Task 1 is cost 2 sec”);
}
public void Run2()
{
Thread.Sleep(3000);
Console.WriteLine(“Task 2 is cost 3 sec”);
}
public void ParallelInvokeMethod()
{
stopWatch.Start();
Parallel.Invoke(Run1, Run2);
stopWatch.Stop();
Console.WriteLine(“Parallel run ” + stopWatch.ElapsedMilliseconds + ” ms.”);//这是在并行的情况下
stopWatch.Restart();
Run1();
Run2();
stopWatch.Stop();
Console.WriteLine(“Normal run ” + stopWatch.ElapsedMilliseconds + ” ms.”);//这是正常情况
}
}
}
分析:正确情况下确实应该是5秒左右,明明白白写着呢,3000+2000=5000ms,而使用了Parallel.Invoke方法只用了3秒,由此可见并行执行提高了很多效率.
扩展2:Parallel.For
这个方法和For循环的功能类似,案例:
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace 并行扩展
{
class Program
{
static void Main(string[] args)
{
Stopwatch stopWatch = new Stopwatch();
stopWatch.Start();
for (int i = 0; i < 10000; i++)
{
for (int j = 0; j < 60000; j++)
{
int sum = 0;
sum += i;
}
}
stopWatch.Stop();
Console.WriteLine(“NormalFor run ” + stopWatch.ElapsedMilliseconds + ” ms.”);
stopWatch.Reset();
stopWatch.Start();
Parallel.For(0, 10000, item =>
{
for (int j = 0; j < 60000; j++)
{
int sum = 0;
sum += item;
}
});
stopWatch.Stop();
Console.WriteLine(“ParallelFor run ” + stopWatch.ElapsedMilliseconds + ” ms.”);
Console.WriteLine(“运行完毕”);
Console.ReadKey();
}
}
}
分析:写了两个循环,做些无用的操作,主要是消耗CPU的资源,通过多次运行结果可以看到们使用了Parallel.For所有的时间差不多也就是单纯for时间的三分之一.是不是在任何时候Parallel.For在任何时候都比这普通的for快呢,肯定不是啊,要是这样的话,C#的for早就淘汰了.
案例:修改代码,添加一个全局变量num
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace 并行扩展
{
class Program
{
static void Main(string[] args)
{
Stopwatch stopWatch = new Stopwatch();
var obj = new Object();
long num = 0;
ConcurrentBag<long> bag = new ConcurrentBag<long>();
stopWatch.Start();
for (int i = 0; i < 10000; i++)
{
for (int j = 0; j < 60000; j++)
{
//int sum = 0;
//sum += item;
num++;
}
}
stopWatch.Stop();
Console.WriteLine(“NormalFor run ” + stopWatch.ElapsedMilliseconds + ” ms.”);
stopWatch.Reset();
stopWatch.Start();
Parallel.For(0, 10000, item =>
{
for (int j = 0; j < 60000; j++)
{
//int sum = 0;
//sum += item;
lock (obj)
{
num++;
}
}
});
stopWatch.Stop();
Console.WriteLine(“ParallelFor run ” + stopWatch.ElapsedMilliseconds + ” ms.”);
//并行的问题,需要多运行几遍,一遍两遍看不出什么东西来
Console.WriteLine(“运行完毕”);
Console.ReadKey();
}
}
}
分析:Parallel.For由于是并行运行的,所以会同时访问访问全局变量num,为了得到正确的结果,要使用lock,所以Parallel.For竟然耗了这么多时间,这是因为并行同时访问全局变量,会出现资源争夺(我的电脑当时开着虚拟机,听着音乐,在程序运行的时候我我Ctrl+S了以下文档,我的电脑出现卡死的情况),大不多的资源都消耗在了资源等待上面.
虽然一直说并说,但一直没有证明一下,你肯定不服,来让你服气:
Parallel.For(0, 100, i =>
{
Console.Write(i + “\t”);
});
Console.WriteLine(“运行完毕”);
从0输出到99，运行后会发现输出的顺序不对，用for顺序肯定是对的，并行同时执行，所以会出现输出顺序不同的情况。
扩展3
Parallel.ForEach
这个方法和普通的foreach很相似,案例:
List<int> list = new List<int>();
for (int i = 0; i < 100000; i++)
{
list.Add(i);
}
Stopwatch stopWatch = new Stopwatch();
stopWatch.Start();
Parallel.ForEach(list, i =>
{
i++;
});
stopWatch.Stop();
Console.WriteLine(stopWatch.ElapsedMilliseconds);
Parallel中途退出循环和异常处理
我们为啥使用Parallel呢,肯定是在处理一些比较好使的任务,当然也很消耗CPU资源和内存,如果我们在中途想停止,怎么办呢?
串行好办啊,一个break搞定,但是串行呢?貌似不知道呢,在并行循环的委托参数中提供了一个ParallelLoopState,该实例提供了Break和Stop方法来帮助我们!
Break: 当然这个是通知并行计算尽快的退出循环，比如并行计算正在迭代100，那么break后程序还会迭代所有小于100的。
Stop：这个就不一样了，比如正在迭代100突然遇到stop，那它啥也不管了，直接退出。
案例:
Stopwatch stopWatch = new Stopwatch();
ConcurrentBag<int> bag = new ConcurrentBag<int>();
stopWatch.Start();
Parallel.For(0, 1000, (i, state) =>
{
if (bag.Count == 300)
{
//state.Stop();//也可能会出现300多,但是不会很夸张,在我的测试下,最多出现303,大部分还是300
state.Break();//如果是break的话,可能会看到Bag count is 300(多),在我多次测试下,得到最夸张的结果是840
return;
}
bag.Add(i);
});
stopWatch.Stop();
Console.WriteLine(“Bag count is ” + bag.Count + “, ” + “一共用时 :”+stopWatch.ElapsedMilliseconds+”ms”);
Console.WriteLine(“运行完毕”);
Console.ReadKey();
异常:
首先任务是并行计算的，处理过程中可能会产生n多的异常，那么如何来获取到这些异常呢？普通的Exception并不能获取到异常，然而为并行诞生的AggregateExcepation就可以获取到一组异常。
这里我们修改Parallel.Invoke的代码，修改后代码如下：
private Stopwatch stopWatch = new Stopwatch();
public void Run1()
{
Thread.Sleep(2000);
Console.WriteLine(“Task 1 is cost 2 sec”);
throw new Exception(“Exception in task 1”);
}
public void Run2()
{
Thread.Sleep(3000);
Console.WriteLine(“Task 2 is cost 3 sec”);
throw new Exception(“Exception in task 2”);
}
public void ParallelInvokeMethod()
{
stopWatch.Start();
try
{
Parallel.Invoke(Run1, Run2);
}
catch (AggregateException aex)
{
foreach (var ex in aex.InnerExceptions)
{
Console.WriteLine(ex.Message);
}
}
stopWatch.Stop();
Console.WriteLine(“Parallel run ” + stopWatch.ElapsedMilliseconds + ” ms.”);
stopWatch.Reset();
stopWatch.Start();
try
{
Run1();
Run2();
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
stopWatch.Stop();
Console.WriteLine(“Normal run ” + stopWatch.ElapsedMilliseconds + ” ms.”);
Console.WriteLine(“运行完毕”);
Console.ReadKey();
