Rx扩展：Parallel.ForEach在哪里？

我有一段使用Parallel.ForEach的代码，可能基于旧版本的Rx扩展或任务并行库。 我安装了当前版本的Rx扩展，但找不到Parallel.ForEach 。 我没有使用该库的任何其他花哨的东西，只是想像这样并行处理一些数据：

 Parallel.ForEach(records, ProcessRecord); 

我发现了这个问题 ，但我不想依赖旧版本的Rx。 但是我无法为Rx找到类似的东西，那么使用当前Rx版本的当前和最直接的方法是什么？ 该项目使用的是.NET 3.5。

Answer 1

如果你有Rx，不需要做所有这些愚蠢的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(IEnumerable source, Action body) { if (source == null) { throw new ArgumentNullException("source"); } if (body == null) { throw new ArgumentNullException("body"); } var items = new List(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(); } }