आरएक्स एक्सटेंशन: समानांतर कहां है। के लिए?

Question

मेरे पास कोड का एक टुकड़ा है जो Parallel.ForEach का उपयोग कर रहा है, शायद आरएक्स एक्सटेंशन या टास्क समांतर लाइब्रेरी के पुराने संस्करण पर आधारित है। मैंने आरएक्स एक्सटेंशन का एक मौजूदा संस्करण स्थापित किया लेकिन Parallel.ForEach नहीं मिला। मैं पुस्तकालय के किसी भी अन्य फैंसी सामान का उपयोग कर रहा है और सिर्फ इस तरह समानांतर में कुछ डेटा की प्रक्रिया करना चाहते हैं:

Parallel.ForEach(records, ProcessRecord); 

मैं this question पाया, लेकिन मैं आरएक्स के एक पुराने संस्करण पर निर्भर नहीं करना चाहते हैं। लेकिन मैं आरएक्स के लिए कुछ ऐसा नहीं ढूंढ पाया, तो वर्तमान आरएक्स संस्करण का उपयोग करने के लिए वर्तमान और सबसे सीधा आगे क्या तरीका है? परियोजना .NET 3.5 का उपयोग कर रही है।

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

(या, यदि आप चाहते हैं पर बनाए रखा आइटम के आदेश:

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

Answer 1

अगर आप आरएक्स है यह सब मूर्ख goosery ऐसा करने के लिए कोई ज़रूरत नहीं:

Answer 2

यहाँ एक सरल प्रतिस्थापन है दक्षता की लागत):

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