Extensions Rx: Où est Parallel.ForEach?

Question

J'ai un morceau de code qui utilise Parallel.ForEach, probablement basé sur une ancienne version des extensions Rx ou la bibliothèque parallèle des tâches. J'ai installé une version actuelle des extensions Rx mais je ne trouve pas Parallel.ForEach. Je ne suis pas d'utiliser tout autre trucs de fantaisie de la bibliothèque et que vous voulez juste pour traiter des données en parallèle comme celui-ci:

Parallel.ForEach(records, ProcessRecord); 

Je trouve this question, mais je ne voudrais pas dépendre d'une ancienne version de Rx. Mais je n'ai pas été en mesure de trouver quelque chose de similaire pour Rx, alors quelle est la manière actuelle et la plus simple de le faire en utilisant une version Rx actuelle? Le projet utilise .NET 3.5.

Answer 1

Pas besoin de faire tout cela goosery stupide si vous avez Rx:

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

(Ou, si vous voulez que l'ordre des éléments maintenus à la coût de l'efficacité):

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

Ou si vous voulez limiter le nombre d'articles en même temps:

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

Answer 2

est ici un simple remplacement:

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(); 
    } 
}