如何优化我的 ExecuteInParallel 函数答案

【问题标题】：How can I optmize my ExecuteInParallel function如何优化我的 ExecuteInParallel 函数
【发布时间】：2013-01-30 11:17:24
【问题描述】：

我已经阅读了很多关于 Parallel .net 4 的内容，我不得不说我有点困惑何时使用它。

这是我的常见场景我的任务是将大量 xml 文件迁移到数据库中。

我通常必须这样做

读取 Xml 文件 (100.000) 等并按数字顺序排列它们（每个文件命名为 1.xml、2.xml 等）。
保存到数据库。

我认为以上是并行编程的完美候选。

从概念上讲，我想一次处理多个文件。

我目前正在这样做：

private ResultEventArgs  progressResults=new ResultEventArgs();

public void ExecuteInParallelTest()
{
    var sw=new Stopwatch();
    sw.Start();
    int index = 0;
    cancelToken = new CancellationTokenSource();
    var parOpts = new ParallelOptions();
    parOpts.CancellationToken = cancelToken.Token;
    parOpts.MaxDegreeOfParallelism = Environment.ProcessorCount;  //It this correct?

    FileInfo[] files = myDirectory.EnumerateFiles("*.xml").ToArray();//Is this faster?
    TotalFiles = files.Count();
    try
    {
        Task t1 = Task.Factory.StartNew(() =>
        {
            try
            {
                Parallel.ForEach(files, parOpts, (file, loopState) =>
                {

                    if (cancelToken.Token.IsCancellationRequested)
                    {
                        cancelToken.Token.ThrowIfCancellationRequested();
                    }

                    index = Interlocked.Increment(ref index);

                    ProcessFile(file,index);

                                progressResults.Status=InProgress                                   

                    OnItemProcessed(TotalFiles,index,etc..);
                });
            }
            catch (OperationCanceledException ex)
            {
                OnOperationCancelled(new progressResults
                    {
                        progressResults.Status=InProgress                               
                        progressResults.TotalCount = TotalFiles;
                        progressResults.FileProcessed= index;
                        //etc..                                  
                    });

            }

            //ContinueWith is used to sync the UI when task completed.
        }, cancelToken.Token).ContinueWith((result) => OnOperationCompleted(new ProcessResultEventArgs
            {
                        progressResults.Status=InProgress
                        progressResults.TotalCount = TotalFiles;
                        progressResults.FileProcessed= index;
                        //etc..
            }), new CancellationTokenSource().Token, TaskContinuationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
    }
    catch (AggregateException ae)
    {
        //TODO:
    }
   }

我的问题：我正在使用 .net 4.0 使用 Parallel 是加快处理这些文件的最佳/更简单的方法。上面的伪代码是否足够好，或者我错过了重要的东西，锁定等......

最重要的问题是：忘记“ProcessFile”，因为我无法对其进行优化，因为我无法控制是否有优化空间

我是否应该将文件分成块，例如 1-1000 - 1001-2000-2001-3000 会提高性能（你是如何做到的）

非常感谢任何回复或链接/代码 sn-p 可以帮助我更好地了解如何改进上述代码。

【问题讨论】：

我建议将此过程流水线化，请参阅this SO post
当你有 IO 操作时，我也会考虑不使用线程。而是使用 Async CTP 和 await 来释放不必要的线程。看看这个很棒的网络广播channel9.msdn.com/Shows/AppFabric-tv/…

标签： c# multithreading task-parallel-library

【解决方案1】：

您没有收到回复的原因是您的代码非常错误。 AsParallel() 对 GetFiles() 没有任何作用，files.Count() 实际上会迭代可枚举，因此不仅您读取文件（或只是目录）两次，而且先执行 Count()，然后再遍历它们将读取文件两次，如果目录被修改，可能会产生不一致的计数。看起来没有必要执行 Task.Factory.StartNew ，因为它是您唯一的任务（在其中产生并行处理）。 Parallel.ForEach 会将所有 OperationCancelledException 封装到单个 AggregateException 中，但只有在所有并行线程完成工作后才会这样做。

【讨论】：

@Andrej tanas 您好，感谢您的评论！非常有价值。这就是为什么我发布这个问题以获得反馈。你能否提供一个代码 sn-p 来说明你将如何重构代码，因为我对你的一些 cmets 以及我将如何解决这些问题有点困惑。对于初学者。我需要报告的总数。至于并行代码，您将如何改进它。谢谢
@Andrej 我在你的回答中也发现有趣的是：你说计数再次迭代。所以我要避免迭代吗？您还提到 GetFiles.AsParallel 什么都不做。为什么？在我的获取文件中有“directoryInfo.EnumerateFiles(pattern).ToArray();”
看到这个：link 关于 IEnumerable.Count() 扩展方法。如果您使用的是 Directory.GetFiles()，请不要使用 Count() 方法，使用返回的字符串数组的 Length 属性。
关于应该如何使用 AsParallel() 的很好的解释可以在这里找到：link
@Andrej 感谢您的链接！但并不是我不知道的事情，只是为了在我获取文件时保持文件的迭代，之后的计数可以忽略不计，并且是 100.000。我已经编辑了我的代码，以便你可以看到完全。我使用 EnumerateFiles 而不是 getFiles，我使用的是计数而不是长度可以更改为长度。我已经反映了 MS 代码，如果有计数，它将返回它，否则迭代。

【解决方案2】：

我保留了代码，因为没有人为我提供合适的答案

【讨论】：