Estensioni Rx: dove si trova Parallel.ForEach?

Question

Ho un pezzo di codice che utilizza Parallel.ForEach, probabilmente basato su una vecchia versione di estensioni Rx o sulla libreria parallela di attività. Ho installato una versione corrente delle estensioni Rx ma non riesco a trovare Parallel.ForEach. Non sto usando qualsiasi altra roba di fantasia della biblioteca e vogliono solo per elaborare alcuni dati in parallelo in questo modo:

Parallel.ForEach(records, ProcessRecord); 

ho trovato this question, ma non vorrei dipendere da una vecchia versione di Rx. Ma non ero in grado di trovare qualcosa di simile per Rx, quindi qual è il modo attuale e più diretto per farlo usando una versione Rx attuale? Il progetto utilizza .NET 3.5.

Answer 1

Non c'è bisogno di fare tutto questo stupido goosery se si dispone di Rx:

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

(O, se si desidera che l'ordine degli elementi mantenuto al costo di efficienza):

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

Oppure, se si desidera limitare il numero di elementi allo stesso tempo:

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

Answer 2

Ecco una semplice sostituzione:

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