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()與orderby
AsOrdered:保留查詢的結果按源序列排序，在並行查詢中，多條數據會被分在多個區域中進行查詢，查詢后再將多個區的數據結果合並到一個結果集中並按源序列順序返回。

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.ElapsedMilliseconds);
        Console.ReadLine();
    }
}
class Product
{
    public string Name { get; set; }
    public string Category { get; set; }
    public int SellPrice { get; set; }
}

 

通過PLINQ執行歸約操作

PLINQ可以簡化對一個序列或者一個組中所有成員應用一個函數的過程，這個過程稱之為歸約操作，如在PLINQ查詢中使用類似於Average,Max,Min,Sum之類的聚合函數就可以充分利用並行所帶來好處。

並行執行的規約和串行執行的規約的執行結果可能會不同，因為在操作不能同時滿足可交換和可傳遞的情況下產生攝入，在每次執行的時候，序列或組中的元素在不同並行任務中分布可能也會有區別，因而在這種操作的情況下可能會產生不同的最終結果，因此，一定要通過對於的串行版本來興義原始的數據源，這樣才能幫助PLINQ獲得最優的執行結果。

下面貼代碼：

class MRESDemo
{
    /*code:釋迦苦僧*/
    static void Main()
    {
        ConcurrentQueue<int> products = new ConcurrentQueue<int>();
        /*向集合中添加多條數據*/
        Parallel.For(0, 6000000, (num) =>
        {
            products.Enqueue(num);
        });

        /*采用LINQ 返回 IEumerable<int>*/
        var productListLinq = (from product in products
                                select product).Average();
        Console.WriteLine("采用Average計算平均值：{0}", productListLinq);

        /*采用PLINQ 返回 ParallelQuery<int>*/
        var productListPLinq = (from product in products.AsParallel()
                                select product).Average();
        Console.WriteLine("采用Average計算平均值：{0}", productListPLinq);
        Console.ReadLine();
    }
}

如上述代碼所示

在LINQ版本中，該方法會返回一個 IEumerable<int>，即調用 Eumerable.Range方法生成指定范圍整數序列的結果，
在PLINQ版本中，該方法會返回一個 ParallelQuery<int>，即調用並行版本中System.Linq.ParallelEumerable的ParallelEumerable.Range方法，通過這種方法得到的結果序列也是並行序列，可以再PLINQ中並行運行。

如果想對特定數據源進行LINQ查詢時，可以定義為  private IEquatable<int> products

如果想對特定數據源進行PLINQ查詢時，可以定義為 private ParallelQuery<int> products

 

並發PLINQ任務

class MRESDemo
{
    /*code:釋迦苦僧*/
    static void Main()
    {
        ConcurrentQueue<Product> products = new ConcurrentQueue<Product>();
        /*向集合中添加多條數據*/
        Parallel.For(0, 600000, (num) =>
        {
            products.Enqueue(new Product() { Category = "Category" + num, Name = "Name" + num, SellPrice = num });
        });
        CancellationTokenSource cts = new CancellationTokenSource();
        /*創建tk1 任務  查詢 符合 條件的數據*/
        Task<ParallelQuery<Product>> tk1 = new Task<ParallelQuery<Product>>((ct) =>
        {
            Console.WriteLine("開始執行 tk1 任務", products.Count);
            Console.WriteLine("tk1 任務中 數據結果集數量為：{0}", products.Count);
            var result = products.AsParallel().Where(p => p.Name.Contains("1") && p.Name.Contains("2"));
            return result;
        }, cts.Token);

        /*創建tk2 任務，在執行tk1任務完成  基於tk1的結果查詢 符合 條件的數據*/
        Task<ParallelQuery<Product>> tk2 = tk1.ContinueWith<ParallelQuery<Product>>((tk) =>
        {
            Console.WriteLine("開始執行 tk2 任務", products.Count);
            Console.WriteLine("tk2 任務中 數據結果集數量為：{0}", tk.Result.Count());
            var result = tk.Result.Where(p => p.Category.Contains("1") && p.Category.Contains("2"));
            return result;
        }, TaskContinuationOptions.OnlyOnRanToCompletion);
        /*創建tk3 任務，在執行tk1任務完成  基於tk1的結果查詢 符合 條件的數據*/
        Task<ParallelQuery<Product>> tk3 = tk1.ContinueWith<ParallelQuery<Product>>((tk) =>
        {
            Console.WriteLine("開始執行 tk3 任務", products.Count);
            Console.WriteLine("tk3 任務中 數據結果集數量為：{0}", tk.Result.Count());
            var result = tk.Result.Where(p => p.SellPrice > 1111 && p.SellPrice < 222222);
            return result;
        }, TaskContinuationOptions.OnlyOnRanToCompletion);

        tk1.Start();

        Task.WaitAll(tk1, tk2, tk3);
        Console.WriteLine("tk2任務結果輸出，篩選后記錄總數為:{0}", tk2.Result.Count());
        Console.WriteLine("tk3任務結果輸出，篩選后記錄總數為:{0}", tk3.Result.Count());

        tk1.Dispose();
        tk2.Dispose();
        tk3.Dispose();
        cts.Dispose();
        Console.ReadLine();
    }
}
class Product
{
    public string Name { get; set; }
    public string Category { get; set; }
    public int SellPrice { get; set; }
}

如代碼所示tk1,tk2,tk3三個任務，tk2,tk3任務的運行需要基於tk1任務的結果，因此，參數中指定了TaskContinuationOptions.OnlyOnRanToCompletion，通過這種方式，每個被串聯的任務都會等待之前的任務完成之后才開始執行，tk2,tk3在tk1執行完成后，這兩個任務的PLINQ查詢可以並行運行，並將會可能地使用多個邏輯內核。

 

取消PLINQ WithCancellation

通過WithCancellation取消當前PLINQ正在執行的查詢操作，代碼如下：

class MRESDemo
{
    /*code:釋迦苦僧*/
    static void Main()
    {
        ConcurrentQueue<Product> products = new ConcurrentQueue<Product>();
        /*向集合中添加多條數據*/
        Parallel.For(0, 600000, (num) =>
        {
            products.Enqueue(new Product() { Category = "Category" + num, Name = "Name" + num, SellPrice = num });
        });
        CancellationTokenSource cts = new CancellationTokenSource();
        CancellationToken token = cts.Token;
        /*創建tk1 任務  查詢 符合 條件的數據*/
        Task<ParallelQuery<Product>> tk1 = new Task<ParallelQuery<Product>>((ct) =>
        {
            var result = products.AsParallel();
            try
            {
                Console.WriteLine("開始執行 tk1 任務", products.Count);
                Console.WriteLine("tk1 任務中 數據結果集數量為：{0}", products.Count);
                result = products.AsParallel().WithCancellation(token).Where(p => p.Name.Contains("1") && p.Name.Contains("2"));

            }
            catch (AggregateException ex)
            {
                foreach (Exception e in ex.InnerExceptions)
                {
                    Console.WriteLine("tk3 錯誤:{0}", e.Message);
                }
            }
            return result;
        }, cts.Token);

        /*創建tk2 任務，在執行tk1任務完成  基於tk1的結果查詢 符合 條件的數據*/
        Task<ParallelQuery<Product>> tk2 = tk1.ContinueWith<ParallelQuery<Product>>((tk) =>
        {
            var result = tk.Result;
            try
            {
                Console.WriteLine("開始執行 tk2 任務", products.Count);
                Console.WriteLine("tk2 任務中 數據結果集數量為：{0}", tk.Result.Count());
                result = tk.Result.WithCancellation(token).Where(p => p.Category.Contains("1") && p.Category.Contains("2"));
            }
            catch (AggregateException ex)
            {
                foreach (Exception e in ex.InnerExceptions)
                {
                    Console.WriteLine("tk3 錯誤:{0}", e.Message);
                }
            }
            return result;
        }, TaskContinuationOptions.OnlyOnRanToCompletion);
        /*創建tk3 任務，在執行tk1任務完成  基於tk1的結果查詢 符合 條件的數據*/
        Task<ParallelQuery<Product>> tk3 = tk1.ContinueWith<ParallelQuery<Product>>((tk) =>
        {
            var result = tk.Result;
            try
            {
                Console.WriteLine("開始執行 tk3 任務", products.Count);
                Console.WriteLine("tk3 任務中 數據結果集數量為：{0}", tk.Result.Count());
                result = tk.Result.WithCancellation(token).Where(p => p.SellPrice > 1111 && p.SellPrice < 222222);
            }
            catch (AggregateException ex)
            {
                foreach (Exception e in ex.InnerExceptions)
                {
                    Console.WriteLine("tk3 錯誤:{0}", e.Message);
                }
            }
            return result;
        }, TaskContinuationOptions.OnlyOnRanToCompletion);

        tk1.Start();

        try
        {
            Thread.Sleep(10);
            cts.Cancel();//取消任務
            Task.WaitAll(tk1, tk2, tk3);

            Console.WriteLine("tk2任務結果輸出，篩選后記錄總數為:{0}", tk2.Result.Count());
            Console.WriteLine("tk3任務結果輸出，篩選后記錄總數為:{0}", tk3.Result.Count());
        }
        catch (AggregateException ex)
        {
            foreach (Exception e in ex.InnerExceptions)
            {
                Console.WriteLine("錯誤:{0}", e.Message);
            }
        }

        tk1.Dispose();
        tk2.Dispose();
        tk3.Dispose();
        cts.Dispose();
        Console.ReadLine();
    }
}
class Product
{
    public string Name { get; set; }
    public string Category { get; set; }
    public int SellPrice { get; set; }
}

 

指定查詢時所需的並行度 WithDegreeOfParallelism

默認情況下，PLINQ總是會試圖利用所有的可用邏輯內核達到最佳性能，在程序中我們可以利用WithDegreeOfParallelism方法指定一個不同最大並行度。

下面貼代碼：

                /*tk1任務 采用所有可用處理器*/
                result = products.AsParallel().WithCancellation(token).WithDegreeOfParallelism(Environment.ProcessorCount).Where(p => p.Name.Contains("1") && p.Name.Contains("2") && p.Category.Contains("1") && p.Category.Contains("2"));
                /*tk1任務 采用1個可用處理器*/
                result = products.AsParallel().WithCancellation(token).WithDegreeOfParallelism(1).Where(p => p.Name.Contains("1") && p.Name.Contains("2") && p.Category.Contains("1") && p.Category.Contains("2"));

好處：如果計算機有8個可用的邏輯內核，PLINQ查詢最多運行4個並發任務，這樣可用使用Parallel.Invoke 加載多個帶有不同並行度的PLINQ查詢，有一些PLINQ查詢的可擴展性有限，因此這些選項可用讓您充分利用額外的內核。

 

使用ForAll 並行遍歷結果

下面貼代碼:

class MRESDemo
{
    /*code:釋迦苦僧*/
    static void Main()
    { 
        ConcurrentQueue<Product> products = new ConcurrentQueue<Product>();
        /*向集合中添加多條數據*/
        Parallel.For(0, 1000, (num) =>
        {
            products.Enqueue(new Product() { Category = "Category" + num, Name = "Name" + num, SellPrice = num });
        });
        products.AsParallel().Where(P => P.Name.Contains("1") && P.Name.Contains("2") && P.Name.Contains("3")).ForAll(product =>
        {
            Console.WriteLine("Name:{0}", product.Name);
        });
         
        Console.ReadLine();
    }
}
class Product
{
    public string Name { get; set; }
    public string Category { get; set; }
    public int SellPrice { get; set; }
}

ForAll是並行，foreach是串行，如果需要以特定的順序處理數據，那么必須使用上述串行循環或方法。

 

WithMergeOptions

通過WithMergeOptions擴展方法提示PLINQ應該優先使用哪種方式合並並行結果片段，如下：

下面貼代碼查看下差異：

class MRESDemo
{
    /*code:釋迦苦僧*/
    static void Main()
    {
        Console.WriteLine("當前計算機處理器數：{0}", Environment.ProcessorCount);
        ConcurrentQueue<Product> products = new ConcurrentQueue<Product>();
        /*向集合中添加多條數據*/
        Parallel.For(0, 600000, (num) =>
        {
            products.Enqueue(new Product() { Category = "Category" + num, Name = "Name" + num, SellPrice = num });
        });
        Stopwatch sw = new Stopwatch();
        Thread.Sleep(1000);
        sw.Restart();
        int count = 0;
        Task tk1 = Task.Factory.StartNew(() =>
        {
            var result = products.AsParallel().WithMergeOptions(ParallelMergeOptions.AutoBuffered).Where(p => p.Name.Contains("1") && p.Name.Contains("2") && p.Category.Contains("1") && p.Category.Contains("2"));
            count = result.Count();
        }); 
        Task.WaitAll(tk1);
        sw.Stop();
        Console.WriteLine("ParallelMergeOptions.AutoBuffered 耗時：{0},數量：{1}", sw.ElapsedMilliseconds, count);

        sw.Restart();
        int count1 = 0;
        Task tk2 = Task.Factory.StartNew(() =>
        {
            var result = products.AsParallel().WithMergeOptions(ParallelMergeOptions.Default).Where(p => p.Name.Contains("1") && p.Name.Contains("2") && p.Category.Contains("1") && p.Category.Contains("2"));
            count1 = result.Count();
        }); 
        Task.WaitAll(tk2);
        sw.Stop();
        Console.WriteLine("ParallelMergeOptions.Default 耗時：{0},數量：{1}", sw.ElapsedMilliseconds, count1);


        sw.Restart();
        int count2 = 0;
        Task tk3 = Task.Factory.StartNew(() =>
        {
            var result = products.AsParallel().WithMergeOptions(ParallelMergeOptions.FullyBuffered).Where(p => p.Name.Contains("1") && p.Name.Contains("2") && p.Category.Contains("1") && p.Category.Contains("2"));
            count2 = result.Count();
        }); 
        Task.WaitAll(tk3);
        sw.Stop();
        Console.WriteLine("ParallelMergeOptions.FullyBuffered 耗時：{0},數量：{1}", sw.ElapsedMilliseconds, count2);


        sw.Restart();
        int count3 = 0;
        Task tk4 = Task.Factory.StartNew(() =>
        {
            var result = products.AsParallel().WithMergeOptions(ParallelMergeOptions.NotBuffered).Where(p => p.Name.Contains("1") && p.Name.Contains("2") && p.Category.Contains("1") && p.Category.Contains("2"));
            count3 = result.Count();
        }); 
        Task.WaitAll(tk4);
        sw.Stop();
        Console.WriteLine("ParallelMergeOptions.NotBuffered 耗時：{0},數量：{1}", sw.ElapsedMilliseconds, count3);

        tk4.Dispose();
        tk3.Dispose();
        tk2.Dispose();
        tk1.Dispose();
        Console.ReadLine();
    }
}
class Product
{
    public string Name { get; set; }
    public string Category { get; set; }
    public int SellPrice { get; set; }
}

需要注意的是：每一個選項都有其優點和缺點，因此一定奧測量顯示第一個結果的時間以及完成整個查詢所需要的時間，這點很重要 。

 

使用PLINQ執行MapReduce算法 ILookup IGrouping

mapreduce ，也稱為Map/reduce 或者Map&Reduce ,是一種非常流行的框架，能夠充分利用並行化處理巨大的數據集，MapReduce的基本思想非常簡單：將數據處理問題分解為以下兩個獨立且可以並行執行的操作：

映射（Map）-對數據源進行操作，為每一個數據項計算出一個鍵值對。運行的結果是一個鍵值對的集合，根據鍵進行分組。

規約（Reduce）-對映射操作產生的根據鍵進行分組的所有鍵值對進行操作，對每一個組執行歸約操作，這個操作可以返回一個或多個值。

下面貼代碼，方便大家理解，但是該案列所展示的並不是一個純粹的MapReduce算法實現：

class MRESDemo
{
    /*code:釋迦苦僧*/
    static void Main()
    {
        ConcurrentQueue<string> list = new ConcurrentQueue<string>();
        list.Enqueue("A");
        list.Enqueue("B");
        list.Enqueue("C");
        list.Enqueue("D");
        list.Enqueue("A");
        list.Enqueue("D");

        Console.WriteLine("Select.......");
        list.AsParallel().Select(p => new
           {
               Name = p,
               Count = 1
           }).ForAll((p) =>
           {
               Console.WriteLine("{0}\t{1}", p.Name, p.Count);
           });

        Console.WriteLine("ILookup.......");
        /*map操作生成的鍵值對由一個單詞和數量1組成，該代碼意在將每個單詞作為鍵並將1作為值加入*/
        ILookup<string, int> map = list.AsParallel().ToLookup(p => p, k => 1);
        foreach (var v in map)
        {
            Console.Write(v.Key);
            foreach (int val in v)
                Console.WriteLine("\t{0}", val);
        }
        /*reduce操作單詞出現的次數*/
        var reduce = from IGrouping<string, int> reduceM in map.AsQueryable()
                     select new
                     {
                         key = reduceM.Key,
                         count = reduceM.Count()
                     };
        Console.WriteLine("IGrouping.......");
        foreach (var v in reduce)
        {
            Console.Write(v.key);
            Console.WriteLine("\t{0}", v.count);
        }

        Console.ReadLine();
    }
}

 

關於PLINQ:聲明式數據並行就寫到這，主要是PLINQ下的查詢注意項和查詢調優的一些擴展方法。如有問題，歡迎指正。

 

作者：釋迦苦僧 出處：http://www.cnblogs.com/woxpp/p/3951096.html 本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接。