并发系列64章(并行编程)第五章

前言

并行编程，先来看下概念。并行编程用于分解计算密集型的任务片段，并将它们分配给多个线程。

划重点，这个是计算密集型的东西，而不是IO密集型。也就是说切割成的片段用于计算使用cpu计算，而不是内存。

如果一个操作时内存密集型那么并行处理是会起反的效果的，因为io意味着等待。原本等待一段的，现在每一段都要等待。

数据的并行处理

class Program
{
	static void Main(string[] args)
	{
		List<int> intlist = new List<int>();
		intlist.Add(1);
		intlist.Add(2);
		intlist.Add(3);
		intlist.Add(4);
		RotateMatrices(intlist);
		Console.ReadKey();
	}
	private static void RotateMatrices(List<int> intlist)
	{
		Parallel.ForEach(intlist, arg =>
		{
			Console.WriteLine(Thread.CurrentThread.ManagedThreadId.ToString());
		});
	}
}

结果是:

上述得出一个结论，不是说并行的每一个都是在不同线程中，而是说并行的任务可能在同一线程。

注意：因为可能在不同线程中，对于公共变量注意锁。

并行聚合

上述并行中，现在有一个需求就是，让他们并行加上某个数，并得出他们的结果。

class Program
{
	static void Main(string[] args)
	{
		List<int> intlist = new List<int>();
		intlist.Add(1);
		intlist.Add(2);
		intlist.Add(3);
		intlist.Add(4);
		var reuslt=RotateMatrices(intlist);
		Console.WriteLine(reuslt);
		Console.ReadKey();
	}
	private static int RotateMatrices(List<int> intlist)
	{
		object mutex = new object();
		int result = 0;
		Parallel.ForEach(source: intlist, localInit: () => 1, body: (item, state, localvalue) =>
			   {

				   Console.WriteLine(Thread.CurrentThread.ManagedThreadId.ToString());
				   return item + localvalue;
		},
		localFinally: localresult =>
		{
			lock (mutex)
			{
				result += localresult;
			}
		}
		);
		return result;
	}
}

localInit: () => 1 设置了为1，然后并行执行了body部分，最后得出结果相加。

从上面可以得出，并行编程其实是阻塞的。如果要达到更好的效果，需要结合异步编程。

上面还有个state 没有用上:

(item, state, localvalue)

这个state 可以 state.stop(); 停止

state.break() 跳出循环。

上面这样计算是有问题的:

class Program
{
	static void Main(string[] args)
	{
		List<int> intlist = new List<int>();
		for (int i = 0; i < 1000; i++)
		{
		intlist.Add(1);
		intlist.Add(2);
		intlist.Add(3);
		intlist.Add(4);
		}
		var reuslt=RotateMatrices(intlist);
		Console.WriteLine("查看result:"+reuslt);
		Console.ReadKey();
	}
	private static int RotateMatrices(List<int> intlist)
	{
		object mutex = new object();
		int result = 0;
		Parallel.ForEach(source: intlist, localInit: () => 1, body: (item, state, localvalue) =>
			   {
				   Console.WriteLine(Thread.CurrentThread.ManagedThreadId.ToString());
				   return item + localvalue;
		},
		localFinally: localvalue =>
		{
			lock (mutex)
			{
				result += localvalue;
			}
		}
		);
		return result;
	}
}

第一次的结果:

第二次的结果:

两次结果不一致。

惊喜不惊喜开心不开心？

刚开始我也很迷茫，后来看了一下群里的大佬点播了一下。

请跑一下下面的，一切都会很清晰的。

private static int RotateMatrices(List<int> intlist)
{
	object mutex = new object();
	int fornumber = 0;
	int tasknumber = 0;
	int result = 0;
	Parallel.ForEach(source: intlist, localInit: () => 1, body: (item, state, localvalue) =>
		   {
			   Console.WriteLine(Thread.CurrentThread.ManagedThreadId.ToString());
			   Console.WriteLine("查看localvalue:"+ localvalue);
			   Console.WriteLine("item:"+item);
			   Interlocked.Add(ref fornumber, 1);
			   return item+localvalue;
		   },
	localFinally: localvalue =>
	{
		Console.WriteLine("查看Tasklocalvalue:" + localvalue);
		lock (mutex)
		{
			result += localvalue-1;
		}
		Interlocked.Add(ref tasknumber, 1);
	}
	);
	Console.WriteLine("fornumber:"+fornumber);
	Console.WriteLine("tasknumber:" + tasknumber);
	return result;
}

重点部分我画了红字:

简化版:

intlist.AsParallel().Sum();

intlist.AsParallel().Aggregate(seed: 0, func: (sum, item) => sum + item);

下一章

我整理了一些:

1.并行调用

2.动态并行

3.并行Linq