Rx拡張機能：Parallel.ForEachはどこですか？

Question

私はParallel.ForEachを使用しているコードを持っています。これはおそらく古いバージョンのRx拡張機能やTasks Parallel Libraryに基づいています。 Rx拡張機能の現在のバージョンをインストールしましたが、Parallel.ForEachが見つかりません。私は、ライブラリの他の派手なものを使用して、ちょうどこのような並列にいくつかのデータを処理したくないよ：

Parallel.ForEach(records, ProcessRecord); 

私はthis questionを見つけたが、私は受信の古いバージョンに依存したくありません。しかし、私はRxに似た何かを見つけることができませんでした。だから、現在のRxバージョンを使用してそれを行うための現在の最も直接的な方法は何ですか？プロジェクトは.NET 3.5を使用しています。

Answer 1

あなたが受信している場合、すべてのこの愚かなgooseryを行う必要がありません：

records.ToObservable() 
    .SelectMany(x => Observable.Start(() => ProcessRecord(x), Scheduler.ThreadPoolScheduler)) 
    .ToList() 
    .First(); 

（または、あなたがしたい場合は項目の順序がで維持が効率のコスト）：

records.ToObservable() 
    .Select(x => Observable.Start(() => ProcessRecord(x), Scheduler.ThreadPoolScheduler)) 
    .Concat() 
    .ToList() 
    .First(); 

または同時にいくつのアイテムを制限したい場合：

records.ToObservable() 
    .Select(x => Observable.Defer(() => Observable.Start(() => ProcessRecord(x), Scheduler.ThreadPoolScheduler))) 
    .Merge(5 /* at a time */) 
    .ToList() 
    .First(); 

Answer 2

ここで簡単に交換です：

class Parallel 
{ 
    public static void ForEach<T>(IEnumerable<T> source, Action<T> body) 
    { 
     if (source == null) 
     { 
      throw new ArgumentNullException("source"); 
     } 
     if (body == null) 
     { 
      throw new ArgumentNullException("body"); 
     } 
     var items = new List<T>(source); 
     var countdown = new CountdownEvent(items.Count); 
     WaitCallback callback = state => 
     { 
      try 
      { 
       body((T)state); 
      } 
      finally 
      { 
       countdown.Signal(); 
      } 
     }; 
     foreach (var item in items) 
     { 
      ThreadPool.QueueUserWorkItem(callback, item); 
     } 
     countdown.Wait(); 
    } 
}