将此 foreach 循环转换为并行执行的正确 PLINQ 语法是什么？

Question

更新于 2011-05-20 12:49AM：foreach 仍然比我的应用程序的并行解决方案快 25%。并且不要使用最大并行度的集合计数，而是使用更接近计算机上核心数量的值。

=

我有一个 IO 密集型任务，我想并行运行。我想对文件夹中的每个文件应用相同的操作。在内部，该操作会产生一个 Dispatcher.Invoke，它将计算出的文件信息添加到 UI 线程上的集合中。因此，从某种意义上说，工作结果是方法调用的副作用，而不是方法调用直接返回的值。

这是我想要并行运行的核心循环

foreach (ShellObject sf in sfcoll)
    ProcessShellObject(sf, curExeName);

该循环的上下文在这里：

        var curExeName = Path.GetFileName(Assembly.GetEntryAssembly().Location);
        using (ShellFileSystemFolder sfcoll = ShellFileSystemFolder.FromFolderPath(_rootPath))
        {
            //This works, but is not parallel.
            foreach (ShellObject sf in sfcoll)
                ProcessShellObject(sf, curExeName);

            //This doesn't work.
            //My attempt at PLINQ.  This code never calls method ProcessShellObject.

            var query = from sf in sfcoll.AsParallel().WithDegreeOfParallelism(sfcoll.Count())
                        let p = ProcessShellObject(sf, curExeName)
                        select p;
        }

    private String ProcessShellObject(ShellObject sf, string curExeName)
    {
        String unusedReturnValueName = sf.ParsingName
        try
        {
            DesktopItem di = new DesktopItem(sf);
            //Up date DesktopItem stuff
            di.PropertyChanged += new PropertyChangedEventHandler(DesktopItem_PropertyChanged);
            ControlWindowHelper.MainWindow.Dispatcher.Invoke(
                (Action)(() => _desktopItemCollection.Add(di)));
        }
        catch (Exception ex)
        {
        }
        return unusedReturnValueName ;
    }

感谢您的帮助！

+汤姆

Answer 1

编辑：关于您的问题的更新。我没有发现该任务是 IO 密集型的 - 并且可能所有文件都来自单个（传统？）磁盘。是的，这会变慢 - 因为您在不可并行的资源中引入了争用，迫使磁盘到处寻找。

有时，IO 密集型任务仍然可以有效地并行化 - 但这取决于资源本身是否可并行化。例如，SSD（寻道时间要短得多）可能完全改变您所看到的特征 - 或者如果您通过网络从多个速度缓慢的服务器获取数据，您可能会遇到 IO -绑定但不在单个通道上。

---

您已经创建了一个查询，但从未使用过它。强制所有内容与查询一起使用的最简单方法是使用 Count() 或 ToList() 或类似的方法。但是，更好的方法是使用 Parallel.ForEach：

var options = new ParallelOptions { MaxDegreeOfParallelism = sfcoll.Count() };
Parallel.ForEach(sfcoll, options, sf => ProcessShellObject(sf, curExeName));

我不确定设置最大并行度是否是正确的方法。它可能有效，但我不确定。解决此问题的另一种方法是将所有操作作为任务启动，并指定TaskCreationOptions.LongRunning。

Answer 2

通过 LINQ 创建的查询对象是 IEnumerable。仅当您枚举它时才会对其进行评估（例如通过 foreach 循环）：

        var query = from sf in sfcoll.AsParallel().WithDegreeOfParallelism(sfcoll.Count())
                    let p = ProcessShellObject(sf, curExeName)
                    select p;
        foreach(var q in query) 
        {
            // ....
        }
        // or:
        var results = query.ToArray(); // also enumerates query

Answer 3

是否应该在最后添加一行

var results = query.ToList();