PLINQ中的分區－互聯網 - 大數據

文章出處

　　最近因為比較忙，好久沒有寫博客了，這篇主要給大家分享一下PLINQ中的分區。上一篇介紹了并行編程，這邊詳細介紹一下并行編程中的分區和自定義分區。

　　先做個假設，假設我們有一個200Mb的文本文件需要讀取，怎么樣才能做到最優的速度呢？對，很顯然就是拆分，把文本文件拆分成很多個小文件，充分利用我們計算機中的多核cpu的優勢，讓每個cpu都充分的利用，達到效率的最大化。然而在PLINQ中也是，我們有一個數據源，如果想進行最大的并行化操作，那么就需要把其拆分為可以多個線程同時訪問的多個部分，這就是PLINQ中的分區。當然，微軟已經為我們想到了這一點，知道他的用戶可能會有這個需求，所以就先說一下微軟給我們提供的默認的一個分區程序吧。

　　微軟提供的默認的分區程序又叫做任務并行庫（TPL），其實就是當你用PLINQ的ForEach的時候，默認在其內部就會給我們進行分區。怎么樣，是不是很方便。不過有時候，你可能會需要自己來進行拆分，那么就是另外一種跟高級一點的用法了，就是PLINQ的自定義分區。自定義分區有兩種，一種是按照范圍分區，另一種是按照區塊分區。其中按照范圍分區在針對鏈表集合能夠提供非常好的性能，比如IList等，不過它也有一點缺點那就是如果一個線程提前完成，它將無法幫助其他線程完成它們的工作。按照區塊分區是當我們不知道我們所要操作的集合的大小的時候，可以使用按照區塊分區，在按區塊分區中，并行循環或查詢中的每個線程或任務都使用一個區塊中一定數量的源元素，對它們進行處理，然后返回檢索其他元素。分區程序可確保分發所有元素，并且沒有重復項。區塊可為任意大小。

　　通常，只有當委托的執行時間為較短到中等程度，源具有大量的元素，并且每個分區的總工作量大致相等時，按范圍分區的速度才會較快。因此，按區塊分區的速度在大多數情況下較快。對于元素數量很少或委托執行時間較長的源，則按區塊分區和按范圍分區的性能大致相等。

　　那么我們如何實現動態分區呢？下面有一個摘自MSDN的示例。

　　每次分區對枚舉器調用 MoveNext 時，枚舉器都會提供包含一個列表元素的分區。對于 PLINQ 和 ForEach，分區是一個 Task 實例。由于請求同時在多個線程上發生，因此對當前索引的訪問是同步的。

  1 //
  2 // An orderable dynamic partitioner for lists
  3 //
  4 class OrderableListPartitioner<TSource> : OrderablePartitioner<TSource>
  5 {
  6     private readonly IList<TSource> m_input;
  7 
  8     public OrderableListPartitioner(IList<TSource> input)
  9         : base(true, false, true)
 10     {
 11         m_input = input;
 12     }
 13 
 14     // Must override to return true.
 15     public override bool SupportsDynamicPartitions
 16     {
 17         get
 18         {
 19             return true;
 20         }
 21     }
 22 
 23     public override IList<IEnumerator<KeyValuePair<long, TSource>>>
 24         GetOrderablePartitions(int partitionCount)
 25     {
 26         var dynamicPartitions = GetOrderableDynamicPartitions();
 27         var partitions =
 28             new IEnumerator<KeyValuePair<long, TSource>>[partitionCount];
 29 
 30         for (int i = 0; i < partitionCount; i++)
 31         {
 32             partitions[i] = dynamicPartitions.GetEnumerator();
 33         }
 34         return partitions;
 35     }
 36 
 37     public override IEnumerable<KeyValuePair<long, TSource>>
 38         GetOrderableDynamicPartitions()
 39     {
 40         return new ListDynamicPartitions(m_input);
 41     }
 42 
 43     private class ListDynamicPartitions
 44         : IEnumerable<KeyValuePair<long, TSource>>
 45     {
 46         private IList<TSource> m_input;
 47         private int m_pos = 0;
 48 
 49         internal ListDynamicPartitions(IList<TSource> input)
 50         {
 51             m_input = input;
 52         }
 53 
 54         public IEnumerator<KeyValuePair<long, TSource>> GetEnumerator()
 55         {
 56             while (true)
 57             {
 58                 // Each task gets the next item in the list. The index is 
 59                 // incremented in a thread-safe manner to avoid races.
 60                 int elemIndex = Interlocked.Increment(ref m_pos) - 1;
 61 
 62                 if (elemIndex >= m_input.Count)
 63                 {
 64                     yield break;
 65                 }
 66 
 67                 yield return new KeyValuePair<long, TSource>(
 68                     elemIndex, m_input[elemIndex]);
 69             }
 70         }
 71 
 72         IEnumerator IEnumerable.GetEnumerator()
 73         {
 74             return
 75                ((IEnumerable<KeyValuePair<long, TSource>>)this)
 76                .GetEnumerator();
 77         }
 78     }
 79 }
 80 
 81 class ConsumerClass
 82 {
 83     static void Main()
 84     {
 85         var nums = Enumerable.Range(0, 10000).ToArray();
 86         OrderableListPartitioner<int> partitioner = new OrderableListPartitioner<int>(nums);
 87 
 88         // Use with Parallel.ForEach
 89         Parallel.ForEach(partitioner, (i) => Console.WriteLine(i));
 90 
 91 
 92         // Use with PLINQ
 93         var query = from num in partitioner.AsParallel()
 94                     where num % 2 == 0
 95                     select num;
 96 
 97         foreach (var v in query)
 98             Console.WriteLine(v);
 99     }
100 }