Extensiones de Rx: ¿Dónde está Parallel.ForEach?

Question

Tengo un código que usa Parallel.ForEach, probablemente basado en una versión anterior de extensiones Rx o en la Biblioteca de tareas paralelas. Instalé una versión actual de las extensiones Rx pero no puedo encontrar Parallel.ForEach. No estoy usando cualquier otra materia de lujo de la biblioteca y sólo quieren procesar algunos datos en paralelo de esta manera:

Parallel.ForEach(records, ProcessRecord); 

He encontrado this question, pero no me gustaría que depender de un viejo versiones de Rx. Pero no pude encontrar algo similar para Rx, entonces, ¿cuál es la forma más sencilla y actual de hacerlo utilizando una versión actual de Rx? El proyecto está usando .NET 3.5.

Answer 1

No hay necesidad de hacer todo esto goosery tonta si tiene Rx:

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

(O bien, si desea que el orden de los elementos mantenidos en el costa de la eficiencia):

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

O si desea limitar el número de productos al mismo tiempo:

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

Answer 2

Aquí hay una simple sustitución:

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