8天玩轉並行開發——第三天 plinq的使用

   

   相信在.net平台下，我們都玩過linq，是的，linq讓我們的程序簡潔優美，簡直玩的是愛不釋手，但是傳統的linq只是串行代碼，在並行的

年代如果linq不支持並行計算那該是多么遺憾的事情啊。

   當然linq有很多種方式，比如linq to sql ,xml，object 等等，如果要將linq做成並行還是很簡單的，這里我就舉一個比較實際一點的例子，

我們知道為了更快的響應用戶操作，碼農們想盡了各種辦法，絞盡了腦汁，其中有一個辦法就是將數據庫數據預加載到內存中，然后通過各種

數據結構的手段來加速CURD，是的，比如一個排序地球人只能做到N(lgN)，那么如果我還想再快一點的話該怎么辦呢？那么現在的並行就能發

揮巨大的優勢，尤其是現在的服務器配置都是在8個硬件線程的情況下，你簡直會狂笑好幾天啊，好，不亂扯了。

 

1：AsParallel(並行化）

下面我們模擬給ConcurrentDictionary灌入1500w條記錄，看看串行和並行效率上的差異，注意我的老爺機是2個硬件線程。

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Collections.Concurrent;
using System.Collections.Generic;

using System.Linq;

class Program
{
    static void Main(string[] args)
    {
        var dic = LoadData();

        Stopwatch watch = new Stopwatch();

        watch.Start();

        //串行執行
        var query1 = (from n in dic.Values
                      where n.Age > 20 && n.Age < 25
                      select n).ToList();

        watch.Stop();

        Console.WriteLine("串行計算耗費時間：{0}", watch.ElapsedMilliseconds);

        watch.Restart();

        var query2 = (from n in dic.Values.AsParallel()
                      where n.Age > 20 && n.Age < 25
                      select n).ToList();

        watch.Stop();

        Console.WriteLine("並行計算耗費時間：{0}", watch.ElapsedMilliseconds);

        Console.Read();
    }

    public static ConcurrentDictionary<int, Student> LoadData()
    {
        ConcurrentDictionary<int, Student> dic = new ConcurrentDictionary<int, Student>();

        //預加載1500w條記錄
        Parallel.For(0, 15000000, (i) =>
        {
            var single = new Student()
            {
                ID = i,
                Name = "hxc" + i,
                Age = i % 151,
                CreateTime = DateTime.Now.AddSeconds(i)
            };
            dic.TryAdd(i, single);
        });

        return dic;
    }

    public class Student
    {
        public int ID { get; set; }

        public string Name { get; set; }

        public int Age { get; set; }

        public DateTime CreateTime { get; set; }
    }
}

執行的結果還是比較震撼的，將近7倍，這是因為plinq的查詢引擎會盡量利用cpu的所有硬件線程。

 

2:常用方法的使用

<1> orderby 

      有時候我們並不是簡單的select一下就ok了，可能需要將結果進行orderby操作，並行化引擎會把要遍歷的數據分區，然后在每個區上進行

orderby操作，最后來一個總的orderby，這里很像算法中的“歸並排序”。

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Collections.Concurrent;
using System.Collections.Generic;

using System.Linq;

class Program
{
    static void Main(string[] args)
    {
        var dic = LoadData();

        var query1 = (from n in dic.Values.AsParallel()
                      where n.Age > 20 && n.Age < 25
                      select n).ToList();


        Console.WriteLine("默認的時間排序如下：");
        query1.Take(10).ToList().ForEach((i) =>
        {
            Console.WriteLine(i.CreateTime);
        });

        var query2 = (from n in dic.Values.AsParallel()
                      where n.Age > 20 && n.Age < 25
                      orderby n.CreateTime descending
                      select n).ToList();

        Console.WriteLine("排序后的時間排序如下：");
        query2.Take(10).ToList().ForEach((i) =>
        {
            Console.WriteLine(i.CreateTime);
        });

        Console.Read();
    }

    public static ConcurrentDictionary<int, Student> LoadData()
    {
        ConcurrentDictionary<int, Student> dic = new ConcurrentDictionary<int, Student>();

        //預加載1500w條記錄
        Parallel.For(0, 15000000, (i) =>
        {
            var single = new Student()
            {
                ID = i,
                Name = "hxc" + i,
                Age = i % 151,
                CreateTime = DateTime.Now.AddSeconds(i)
            };
            dic.TryAdd(i, single);
        });

        return dic;
    }

    public class Student
    {
        public int ID { get; set; }

        public string Name { get; set; }

        public int Age { get; set; }

        public DateTime CreateTime { get; set; }
    }
}

 

<2> sum(),average()等等這些聚合函數的效果跟orderby類型一樣，都是實現了類型歸並排序的效果，這里就不舉例子了。

 

3：指定並行度，這個我在前面文章也說過，為了不讓並行計算占用全部的硬件線程，或許可能要留一個線程做其他事情。

        var query2 = (from n in dic.Values.AsParallel()
                      .WithDegreeOfParallelism(Environment.ProcessorCount - 1)
                      where n.Age > 20 && n.Age < 25
                      orderby n.CreateTime descending
                      select n).ToList();

 

4： 了解ParallelEnumerable類

   首先這個類是Enumerable的並行版本，提供了很多用於查詢實現的一組方法，截個圖，大家看看是不是很熟悉，要記住，他們都是並行的。

下面列舉幾個簡單的例子。

class Program
{
    static void Main(string[] args)
    {
        ConcurrentBag<int> bag = new ConcurrentBag<int>();

        var list = ParallelEnumerable.Range(0, 10000);

        list.ForAll((i) =>
        {
            bag.Add(i);
        });

        Console.WriteLine("bag集合中元素個數有:{0}", bag.Count);

        Console.WriteLine("list集合中元素個數總和為:{0}", list.Sum());

        Console.WriteLine("list集合中元素最大值為:{0}", list.Max());

        Console.WriteLine("list集合中元素第一個元素為:{0}", list.FirstOrDefault());

        Console.Read();
    }
}

 

5： plinq實現MapReduce算法

  mapReduce是一個非常流行的編程模型，用於大規模數據集的並行計算，非常的牛X啊，記得mongodb中就用到了這個玩意。

map：  也就是“映射”操作，可以為每一個數據項建立一個鍵值對，映射完后會形成一個鍵值對的集合。

reduce：“化簡”操作，我們對這些巨大的“鍵值對集合“進行分組，統計等等。

具體大家可以看看百科：http://baike.baidu.com/view/2902.htm

 

下面我舉個例子，用Mapreduce來實現一個對age的分組統計。

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

class Program
{
    static void Main(string[] args)
    {
        List<Student> list = new List<Student>()
        {
            new Student(){ ID=1, Name="jack", Age=20},
            new Student(){ ID=1, Name="mary", Age=25},
            new Student(){ ID=1, Name="joe", Age=29},
            new Student(){ ID=1, Name="Aaron", Age=25},
        };

        //這里我們會對age建立一組鍵值對
        var map = list.AsParallel().ToLookup(i => i.Age, count => 1);

        //化簡統計
        var reduce = from IGrouping<int, int> singleMap
                     in map.AsParallel()
                     select new
                     {
                         Age = singleMap.Key,
                         Count = singleMap.Count()
                     };

        ///最后遍歷
        reduce.ForAll(i =>
        {
            Console.WriteLine("當前Age={0}的人數有:{1}人", i.Age, i.Count);
        });
    }

    public class Student
    {
        public int ID { get; set; }

        public string Name { get; set; }

        public int Age { get; set; }

        public DateTime CreateTime { get; set; }
    }
}