Parallel的使用之Parallel.forrech

首先一般Foreach，Parallel.For,Parallel.Foreach三种情况的效率问题，

要知道的是：Parallel.Foreach性能最优，其次 Parallel.For 最后 Foreach 但是这只是一般情况下，如果循环里面执行的代码块非常简单，执行时间特别短，那么性能最优的是Foreach ，原因很简单，因为 Parallel是需要创建线程的，会开销线程耗时间。

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using System.Threading.Tasks;

namespace 测试ParallelFor
{
class Program
{
/*
* 测试分析结果
* Parallel.For、Parallel.Foreach发挥出了平行运算的优势，将效率提高了接近一半左右。
*
* 测试总结
* 对于Parallel.For、Parallel.Foreach的使用应该要特别小心，
* 它们的优势是处理列表很长，且对列表内的元素进行很复杂的业务逻辑，且不会使用共享资源，
* 只针对自身的业务逻辑处理，方才能提升效率。
* 因为如果逻辑过于简单的话，创建线程的花费将大于业务执行的花费，得不偿失。
*/
static void Main(string[] args)
{
//产生测试资料
List<int> testData = new List<int>();
Random Rand = new Random();
//产生乱数列表
for (int i = 0; i < 1000000; i++)
{
testData.Add(Rand.Next(1000));
}
//打印正确结果
Console.WriteLine(testData.Sum());

for (int i = 0; i < 5; i++)
{
Console.WriteLine();
TestFor(testData);
TestParallelFor(testData);
TestParallelForeach(testData);
}
Console.ReadKey();
}

static void TestFor(List<int> testData)
{
DateTime time1 = DateTime.Now;
foreach (var item in testData)
{
item.ToString();
}
Console.WriteLine(string.Format("ForEach: t{0} in {1}", testData.Sum(), (DateTime.Now - time1).TotalMilliseconds));
}

static void TestParallelFor(List<int> testData)
{
DateTime time1 = DateTime.Now;
Parallel.For(0, testData.Count, (i, loopState) =>
{
testData[i].ToString();
});
Console.WriteLine(string.Format("Parallel.For: t{0} in {1}", testData.Sum(), (DateTime.Now - time1).TotalMilliseconds));
}

static void TestParallelForeach(List<int> testData)
{
//记录结果用
DateTime time1 = DateTime.Now;
Parallel.ForEach(testData, (item, loopState) =>
{
item.ToString();
});
Console.WriteLine(string.Format("Parallel.ForEach:t{0} in {1}", testData.Sum(), (DateTime.Now - time1).TotalMilliseconds));
}
}
}

实验2：Parallel的线程管理情况

测试结果：

Parallel会再最近的一个thread结束后，把该完成的ThreadId作为新的开辟的线程的Id,

源码：

实验3：最大线程数。线程是CPU进行调度的单位，进程是系统进程调度的单位。线程组成进程。设置平行运行最大最大线程数：2个；这样系统运行可控，不会造成高CPU的情况

结果及源码：

arallel的跳出循环以及终止循环。

Parallel.ForEach(list,new ParallelOptions(){ MaxDegreeOfParallelism=2}, (p, state1) =>
{
Invoke(p);
state1.Break();//Break用于根据条件过滤循环，Break不是Continue，不要搞混了！Break是执行完现有的迭代后跳出！
state1.Stop();//Stop方法用于退出Paraller循环,表示立刻退循环,cease=终止
return; //注意：不论是Break还是Stop方法，后面的return语句是必须的，否则当前循环体第13行的语句还是会被执行。
});

ParallelLoopState.Stop() 提供了退出循环的方法，这种方式要比其他两种方法更快。这个方法通知循环不要再启动执行新的迭代，并尽可能快的推出循环。

ParallelLoopState.IsStopped 属性可用来判定其他迭代是否调用了 Stop 方法。

Break不是Continue，不要搞混了！Break是执行完现有的迭代后跳出！

ParallelLoopState.Break() 通知循环继续执行本元素前的迭代，但不执行本元素之后的迭代。最前调用 Break 的起作用，并被记录到 ParallelLoopState.LowestBreakIteration 属性中

其他知识点：

未设置最大线程数的情况下:

1.为设置最大线程的情况下，TPL默认线程数为任务数(系统允许的情况下，设置ThreadPool.SetMaxThreads没有效果)。

2.TPL默认启动5个线程，任务数小于5的话，启动任务数个线程。

3.如果任务较多,TPL在初始化5个线程后，每隔100毫秒左右新增线程，直到达到最大线程数。如果新增线程的过程中有任务完成，那么就不会新增线程。

缺点：线程数无法控制，容易造成高CPU，系统失去响应。

举一个比较有实际意义的案例：

在数据库有会员4000万左右，现在需要把这些会员全部获取到，并且根据会员卡号查询其他相关表信息，来给各个会员打上标签，来指定一些分类营销例如：给不用会员的使用习惯发不同的优惠券，推送短信，公众号推广，积分提醒等等

如果我们用一个主线程直接for循环，由于每一个会员的属性都特别复杂，估计全部跑完要一个月时间，但是我们取得数据都是根据会员动态信息来分析的，当你全部完成打标签时，已打标签跟会员当前的特性已经不完全吻合了，那已经是一个月的数据，而且单线程这样跑，会出现很多意料之外的问题，所以这个方案显然是不满足需求的。

那么只能用多线程了。但是多线程方式很多，可以用 Parallel.ForEach

思路很简单：会员里面是有主键ID的，我们这边是int类型，如果表设计的时候不是int类型，可以考虑其他字段，只要是唯一的，可排序的就可以，不一定是id。

1：查询数据库会员总数量

2：判断一下大概每个线程一次执行多少数量，例如200

3：那么大概遍历次数： int Labercount = (int)Math.Ceiling(Convert.ToDecimal(dtLabercount.Rows[0][0]) / 200);

4：定义3个数组，listobj是数组里面的数组，目的：subItemList需要放2个参数，第一个参数就是顺序，第二个是对低多少条数据，

var listObj = new List<List<string>>();
List<string> subItemList;

for (int i = Labercount; i < count; i++)
{
subItemList .Add((i).ToString());
subItemList .Add((200 + 200* i).ToString());

listObj.Add(subItemList);
subItemList .Clear();
}

这样 listObj里面的数量就是 4000万/200个数组

var option = new ParallelOptions { MaxDegreeOfParallelism = thread}; // 定义一个最大线程数：

var watch = Stopwatch.StartNew();

System.Threading.Tasks.Parallel.ForEach(listObj.AsParallel(), option, (parameter) =>
{
try
{
Console.WriteLine("时间： " + DateTime.Now + " 会员信息ID号为:" + Convert.ToInt32(parameter[0]) * 200 + " 段号数据");

//构造传递的参数
Hashtable hs = new Hashtable();
hs.Add("pageIndex", parameter[0]);
hs.Add("pageSize", 200);

DataTable dt = member.QueryMember(hs);

if (dt != null && dt.Rows.Count > 0)
{
--- insert 到分类表
DataTable dtdetail = GetMemberCategorydt(dt);
int result = cate.BulkCopyDataToDB(dtdetail);
}
}
catch (Exception ex)
{
//Logger.WriteAppError(ex, parameter[0] + "号段报错");
Console.WriteLine(ex+parameter[0] + "号段报错"+ ex.Message);
}
});

watch.Stop();
Console.WriteLine("更新记录完成,用时{0}\r\n", watch.Elapsed.TotalMinutes);

这里需要理解Parallel.ForEach里面的参数：

1：第一个参数是数组，就是上面代码分析后得到的数组对象，每个里面都有会员的主键信息

2：第二个参数是设置线程最大数，如果不设置最大数，就会默认先创建5个，然后不停的新增线程数，知道最大为止，这是不可取的，因为并不是线程越多就越好。

3： (parameter) =>是拉姆达表达式的写法，里面就是 listObj数组的元素，当然这里的listObj里面还是数组，如果前面设计的是不那么复杂，直接是string或者int类型数据，那么parameter也就是string 或者int类型的数据。

4：根据这个数据去会员表查询，这里做排序查询，相当于分页查询一下，根据parameter里面的参数，做查询，只是这里的分页不是排序的，而是无序的。这里每次查询就是小于等于200个会员信息。为什么是小于等于，因为我们无法保证里面的参数每一条在数据库都存在，

5：获取到会员卡号。根据会员卡号到其他关联表取数据分别做不同的分析

整个流程下来，4000万会员大概在3天左右全部跑完。勉强可以接受

再添加一个比较好理解的代码

class ParallelForeach
{
public void TestParllerlForeach()
{
var list = new List<int>(100);
for (int i = 0; i < 100; i++)
{
list.Add(i);
}

// 这里的p是遍历list的数据 ,但是不是排序的取，而且随机的取出， MaxDegreeOfParallelism是最大线程数量

Parallel.ForEach(list,new ParallelOptions() {MaxDegreeOfParallelism=2 }, (p) =>
{
InvokoForeach(p);
});
}

private void InvokoForeach(int i)
{
Console.WriteLine("------>当前ThreadId：" + Thread.CurrentThread.ManagedThreadId);
Thread.Sleep(1000);
Console.WriteLine("<-------ThreadId:" + Thread.CurrentThread.ManagedThreadId + "完毕");
}
}