Parallel的基本使用
private static void Run1() { Console.WriteLine("任务1:3s"); Thread.Sleep(3000); } private static void Run2() { Console.WriteLine("任务2:5s"); Thread.Sleep(5000); } private static void ParallerTest01() { var watch = Stopwatch.StartNew(); watch.Start(); Run1(); Run2(); Console.WriteLine("串行耗时:{0}", watch.ElapsedMilliseconds); watch.Restart(); Parallel.Invoke(Run1, Run2); Console.WriteLine("并行耗时:{0}", watch.ElapsedMilliseconds); Console.Read(); } private static void ParallerTest02() { for (var j = 1; j < 5; j++) { Console.WriteLine("第{0}次循环", j); #region 并行 var bag = new ConcurrentBag<int>(); var watch = Stopwatch.StartNew(); watch.Start(); Parallel.For(0, 20_000_000, i => { bag.Add(i); }); Console.WriteLine($"并行耗时:{watch.ElapsedMilliseconds}"); #endregion #region 串行 var bag1 = new ConcurrentBag<int>(); watch = Stopwatch.StartNew(); watch.Start(); for (var i = 0; i < 20_000_000; i++) { bag1.Add(i); } Console.WriteLine($"串行耗时:{watch.ElapsedMilliseconds}"); #endregion } }Parallel在linq中使用
public partial class Program { private static void 01() { var students = new ConcurrentDictionary<int, Student>(); // 并行生成1000万条数据 Parallel.For(0, 10_000_000, (i) => { var single = new Student { Id = i, Name = "name" + i, Age = i % 100, CreateTime = DateTime.Now.AddSeconds(i) }; students.TryAdd(i, single); }); Console.WriteLine("数据已生成"); // PLINQ查询 var watch = new Stopwatch(); watch.Start(); var query1 = (from n in students.Values.AsParallel() where n.Age > 18 && n.Age < 80 select n).ToList(); watch.Stop(); Console.WriteLine("PLINQ耗时：{0}", watch.ElapsedMilliseconds); // LINQ查询 watch.Restart(); var query2 = (from n in students.Values where n.Age > 18 && n.Age < 80 select n).ToList(); watch.Stop(); Console.WriteLine("LINQ耗时：{0}", watch.ElapsedMilliseconds); } private static void 02() { var students = new ConcurrentDictionary<int, Student>(); Parallel.For(0, 10, (i) => { var single = new Student { Id = i, Name = "name" + i, Age = i % 100, CreateTime = DateTime.Now.AddSeconds(i) }; students.TryAdd(i, single); }); Console.WriteLine("数据生成结束"); // 测试PLNQ与LINQ查询 var query = from n in students.Values.AsParallel() select new {Name = n.Name, ThreadId = Thread.CurrentThread.ManagedThreadId}; foreach (var o in query) { Console.WriteLine($"{o} - from ThreadId = {Thread.CurrentThread.ManagedThreadId}"); } Console.WriteLine($"CurrentThreadId:{Thread.CurrentThread.ManagedThreadId}"); } }PLINQ 执行MapReduce算法 基本信息
MapReduce(映射和规约)也称 Map/Reduce 或 Map & Reduce，充分运用并行方式处理大数据集。基本思想是将数据处理问题分解为两个独立的且可并行执行的操作： Map 和 Reduce。
Map：对数据源进行操作，为每个数据项计算出一个键值，运行的结果是一个键-值对的集合，并且根据键分组。 Reduce：对Map产生的键-值对进行操作，对每个组进行规约操作，返回结果值(一个或多个)。
这里写图片描述
程序示例： 一段文本，统计其中单词的个数，数据量很少，只是演示PLINQ关于 MapReduce的执行方式。
第一步：建立了一个 单词 和 数字1 组成的键-值对。这里使用的是ILookup<>接口。 详细参考：https://msdn.microsoft.com/zh-cn/library/bb534291(v=VS.100).aspx
第二步：对键值对进行分组 select操作，并且选择出Count大于1的word。
using System; using System.Collections.Generic; using System.Linq; using System.Text; namespace Sample6_7_plinq_mapreduce { class Program { public static string strTarget = @"English is a West Germanic language that was first spoken in early medieval England and is now a global lingua franca. It is an official language of almost 60 sovereign states, the most commonly spoken language in the United Kingdom, the United States, Canada, Australia, Ireland, and New Zealand, and a widely spoken language in countries in the Caribbean, Africa, and southeast Asia. It is the third most common native language in the world, after Mandarin and Spanish. It is widely learned as a second language and is an official language of the United Nations, of the European Union, and of many other world and regional international organisations."; static void Main(string[] args) { string[] words = strTarget.Split(' '); ILookup<string, int> map = words.AsParallel().ToLookup(p => p, k => 1); var reduce = from IGrouping<string, int> wordMap in map.AsParallel() where wordMap.Count() > 1 select new { Word = wordMap.Key, Count = wordMap.Count() }; foreach (var word in reduce) Console.WriteLine("Word: '{0}' : Count: {1}", word.Word, word.Count); Console.ReadLine(); } } }