C#并行编程-PLINQ:声明式数据并行

目录

• C#并行编程-相关概念

• C#并行编程-Parallel

• C#并行编程-Task

• C#并行编程-并发集合

• C#并行编程-线程同步原语

• C#并行编程-PLINQ:声明式数据并行

背景

通过LINQ可以方便的查询并处理不同的数据源，使用Parallel LINQ (PLINQ)来充分获得并行化所带来的优势。

PLINQ不仅实现了完整的LINQ操作符，而且还添加了一些用于执行并行的操作符，与对应的LINQ相比，通过PLINQ可以获得明显的加速，但是具体的加速效果还要取决于具体的场景，不过在并行化的情况下一段会加速。

如果一个查询涉及到大量的计算和内存密集型操作，而且顺序并不重要，那么加速会非常明显，然而，如果顺序很重要，那么加速就会受到影响。

asparallel() 启用查询的并行化

下面贴代码，看下效果，详情见注释：

class MRESDemo{    /*code:释迦苦僧*/    static void Main()    {        ConcurrentQueue<PRoduct> products = new ConcurrentQueue<Product>();        /*向集合中添加多条数据  可以修改数据量查看Linq和Plinq的性能*/        Parallel.For(0, 600000, (num) =>        {            products.Enqueue(new Product() { Category = "Category" + num, Name = "Name" + num, SellPrice = num });        });        /*采用LINQ查询符合条件的数据*/        Stopwatch sw = new Stopwatch();        sw.Restart();        var productListLinq = from product in products                              where (product.Name.Contains("1") && product.Name.Contains("2") && product.Category.Contains("1") && product.Category.Contains("2"))                              select product;        Console.WriteLine("采用Linq 查询得出数量为:{0}", productListLinq.Count());        sw.Stop();        Console.WriteLine("采用Linq 耗时:{0}", sw.ElapsedMilliseconds);        /*采用PLINQ查询符合条件的数据*/        sw.Restart();        var productListPLinq = from product in products.AsParallel() /*AsParallel 试图利用运行时所有可用的逻辑内核，从而使运行的速度比串行的版本要快 但是需要注意开销所带来的性能损耗*/                               where (product.Name.Contains("1") && product.Name.Contains("2") && product.Category.Contains("1") && product.Category.Contains("2"))                               select product;        Console.WriteLine("采用PLinq 查询得出数量为:{0}", productListPLinq.Count());        sw.Stop();        Console.WriteLine("采用PLinq 耗时:{0}", sw.ElapsedMilliseconds);        Console.ReadLine();    }}class Product{    public string Name { get; set; }    public string Category { get; set; }    public int SellPrice { get; set; }}

当前模拟的数据量比较少，数据量越多，采用并行化查询的效果越明显

AsOrdered()与orderbyAsOrdered:保留查询的结果按源序列排序，在并行查询中，多条数据会被分在多个区域中进行查询，查询后再将多个区的数据结果合并到一个结果集中并按源序列顺序返回。

orderby：将返回的结果集按指定顺序进行排序

下面贴代码方便大家理解：

class MRESDemo{    /*code:释迦苦僧*/    static void Main()    {        ConcurrentQueue<string> products = new ConcurrentQueue<string>();        products.Enqueue("E");        products.Enqueue("F");        products.Enqueue("B");        products.Enqueue("G");        products.Enqueue("A");        products.Enqueue("C");        products.Enqueue("SS");        products.Enqueue("D");        /*不采用并行化  其数据输出结果  不做任何处理   */        var productListLinq = from product in products                               where (product.Length == 1)                               select product;        string appendStr = string.Empty;        foreach (string str in productListLinq)        {            appendStr += str + " ";        }        Console.WriteLine("不采用并行化 输出:{0}", appendStr);        /*不采用任何排序策略  其数据输出结果 是直接将分区数据结果合并起来 不做任何处理   */        var productListPLinq = from product in products.AsParallel()                               where (product.Length == 1)                               select product;         appendStr = string.Empty;        foreach (string str in productListPLinq)        {            appendStr += str + " ";        }        Console.WriteLine("不采用AsOrdered 输出:{0}", appendStr);        /*采用 AsOrdered 排序策略  其数据输出结果 是直接将分区数据结果合并起来 并按原始数据顺序排序*/        var productListPLinq1 = from product in products.AsParallel().AsOrdered()                                where (product.Length == 1)                                select product;        appendStr = string.Empty;        foreach (string str in productListPLinq1)        {            appendStr += str + " ";        }        Console.WriteLine("采用AsOrdered 输出:{0}", appendStr);        /*采用 orderby 排序策略  其数据输出结果 是直接将分区数据结果合并起来 并按orderby要求进行排序*/        var productListPLinq2 = from product in products.AsParallel()                                where (product.Length == 1)                                orderby product                                select product;        appendStr = string.Empty;        foreach (string str in productListPLinq2)        {            appendStr += str + " ";        }        Console.WriteLine("采用orderby 输出:{0}", appendStr);        Console.ReadLine();    }}

在PLINQ查询中，AsOrdered()和orderby子句都会降低运行速度，所以如果顺序并不是必须的，那么在请求特定顺序的结果之前，将加速效果与串行执行的性能进行比较是非常重要的。

指定执行模式 WithExecutionMode

对串行化代码进行并行化，会带来一定的额外开销，Plinq查询执行并行化也是如此，在默认情况下，执行PLINQ查询的时候，.NET机制会尽量避免高开销的并行化算法，这些算法有可能会将执行的性能降低到地狱串行执行的性能。

.NET会根据查询的形态做出决策，并不开了数据集大小和委托执行的时间，不过也可以强制并行执行，而不用考虑执行引擎分析的结果，可以调用WithExecutionMode方法来进行设置。、

下面贴代码，方便大家理解

class MRESDemo{    /*code:释迦苦僧*/    static void Main()    {        ConcurrentQueue<Product> products = new ConcurrentQueue<Product>();        /*向集合中添加多条数据*/        Parallel.For(0, 6000000, (num) =>        {            products.Enqueue(new Product() { Category = "Category" + num, Name = "Name" + num, SellPrice = num });        });        /*采用并行化整个查询 查询符合条件的数据*/        Stopwatch sw = new Stopwatch();        sw.Restart();        var productListLinq = from product in products.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism)                              where (product.Name.Contains("1") && product.Name.Contains("2") && product.Category.Contains("1") && product.Category.Contains("2"))                              select product;        Console.WriteLine("采用并行化整个查询 查询得出数量为:{0}", productListLinq.Count());        sw.Stop();        Console.WriteLine("采用并行化整个查询 耗时:{0}", sw.ElapsedMilliseconds);        /*采用默认设置 由.NET进行决策 查询符合条件的数据*/        sw.Restart();        var productListPLinq = from product in products.AsParallel().WithExecutionMode(ParallelExecutionMode.Default)                               where (product.Name.Contains("1") && product.Name.Contains("2") && product.Category.Contains("1") && product.Category.Contains("2"))                               select product;        Console.WriteLine("采用默认设置 由.NET进行决策 查询得出数量为:{0}", productListPLinq.Count());        sw.Stop();        Console.WriteLine("采用默认设置 由.NET进行决策 耗时:{0}", sw.ElapsedMillise