Parallel.ForEach 问题答案

【问题标题】：Parallel.ForEach questionsParallel.ForEach 问题
【发布时间】：2011-05-17 21:03:39
【问题描述】：

我在 C#/VS2010 中使用 Parallel.ForEach 循环进行处理，我有几个问题。

首先，我有一个进程需要从远程 Web 服务中提取信息，然后需要动态构建图像 (GDI)。

我有一个将所有功能封装到单个对象中的类，该对象具有两个主要方法 Load() 和 CreateImage()，所有 GDI 管理/WebRequests 在该对象内“黑盒化”。

然后我创建一个包含所有需要处理的对象的 GenericList，并使用以下代码遍历该列表：

try
        {
            Parallel.ForEach(MyLGenericList, ParallelOptions, (MyObject, loopState) =>
            {                                       

                    MyObject.DoLoad();
                    MyObject.CreateImage();
                    MyObject.Dispose();

                if (loopState.ShouldExitCurrentIteration || loopState.IsExceptional)
                    loopState.Stop();
            });
        }
        catch (OperationCanceledException ex)
        {
            // Cancel here
        }
        catch (Exception ex)
        {
            throw ex;
        }

现在我的问题是：

鉴于列表中可能有一万个要解析的项目，以上代码是解决此问题的最佳方法吗？欢迎任何其他想法
我有一个问题，当我开始进程时，对象被创建/加载并且图像创建速度非常快，但是在大约 600 个对象之后，进程开始爬行。它最终没有完成，这正常吗？

在此先感谢 :) 亚当

【问题讨论】：

拥有一个除了抛出它捕获的异常什么都不做的catch块有什么意义？
嗨，这足以显示我在 May 应用程序中所做的事情，显然没有人对我的错误处理感兴趣，因为它与问题无关 :)
很公平，但我会完全省略 catch 块（显然是第二个）——尤其是因为你只是在画一个例子。

标签： c# .net multithreading task-parallel-library

【解决方案1】：

我不确定并行下载数据是个好主意，因为它会阻塞很多线程。将您的任务拆分为生产者和消费者。然后您可以分别并行化它们中的每一个。

这是一个单一生产者和多个消费者的示例。
（如果消费者比生产者快，你可以使用普通的 foreach 而不是 parallel.ForEach）

var sources = BlockingCollection<SourceData>();
var producer = Task.Factory.CreateNew(
    () => {
        foreach (var item in MyGenericList) {
            var data = webservice.FetchData(item);
            sources.Add(data)
        }
        sources.CompleteAdding();
    }
)
Parallel.ForEach(sources.GetConsumingPartitioner(),
                 data => {
                     imageCreator.CreateImage(data);
                 });

（GetConsumingPartitioner 扩展是ParallelExtensionsExtras 的一部分）

编辑更完整的例子

var sources = BlockingCollection<SourceData>();

var producerOptions = new ParallelOptions { MaxDegreeOfParallelism = 5 };
var consumerOptions = new ParallelOptions { MaxDegreeOfParallelism = -1 };

var producers = Task.Factory.CreateNew(
    () => {
        Parallel.ForEach(MyLGenericList, producerOptions, 
            myObject => {
                myObject.DoLoad()
                sources.Add(myObject)
            });
        sources.CompleteAdding();
    });
Parallel.ForEach(sources.GetConsumingPartitioner(), consumerOptions,
    myObject => {
        myObject.CreateImage();
        myObject.Dispose();
    });

使用此代码，您可以优化并行下载量，同时让 cpu 忙于图像处理。

【讨论】：

您好，感谢您的帮助。在上面的示例中，我必须等到从 web 服务中检索到每个项目，这可能需要一段时间。我希望填充我的对象/在一个单独的线程中一次创建五个图像，因为它通过列表以加快速度。
不，该示例有一个线程不断检索项目。一旦获取了一个项目，消费者部分就开始处理它（同时获取下一个项目）。如果您认为同时获取五个项目更快，请使用多个生产者。（我会更新例子）
再次感谢您的帮助；我遇到的一个问题是 ParallelExtensionsExtras.dll，我已经下载了它，添加了参考，但 GetConsumingPartitioner 似乎从未出现。我得到 GetConsumingEnumerable 没问题。有什么想法吗？
好的，谢谢；这似乎是正确的方法codeParallel.ForEach(BlockingCollectionExtensions.GetConsumingPartitioner(sources), consumerOptions, myObject => {

【解决方案2】：

当循环体所做的工作受 CPU 限制时，使用默认设置的 Parallel.ForEach 方法效果最佳。如果你同步阻塞或将工作交给另一方，调度器会认为 CPU 仍然不忙，继续塞进更多的任务，努力使用系统中的所有 CPU。

在您的情况下，您只需选择合理数量的重叠下载以并行发生，并在您的 ForEach 选项中设置该值，因为您的循环不会使 CPU 饱和。

【讨论】：