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