Parallel.ForEach 在迭代结束时表现得像一个常规的每个

Question

当我运行这样的事情时遇到了这个问题：

Parallel.ForEach(dataTable.AsEnumerable(), row =>
{
   //do processing
}

假设有 500 多条记录说 870。一旦 Parallel.ForEach 完成 850，它似乎是按顺序运行的，即一次只有 1 个操作。它非常快地完成了 850 次操作，但是当它接近迭代结束时，它变得非常慢，并且似乎每个操作都像常规操作一样执行。我什至尝试了 2000 条记录。

我的代码有问题吗？请提出建议。

下面是我正在使用的代码

抱歉，我刚刚发布了错误的示例。这是正确的代码：

Task newTask = Task.Factory.StartNew(() =>
{
    Parallel.ForEach(dtResult.AsEnumerable(), dr =>
    {
        string extractQuery = "";
        string downLoadFileFullName = "";
        lock (foreachObject)
        {

            string fileName = extractorConfig.EncodeFileName(dr);
            extractQuery = extractorConfig.GetExtractQuery(dr);
            if (string.IsNullOrEmpty(extractQuery)) throw new Exception("Extract Query not found. Please check the configuration");

            string newDownLoadPath = CommonUtil.GetFormalizedDataPath(sDownLoadPath, uKey.CobDate);
            //create folder if it doesn't exist
            if (!Directory.Exists(newDownLoadPath)) Directory.CreateDirectory(newDownLoadPath);
            downLoadFileFullName = Path.Combine(newDownLoadPath, fileName);
        }
        Interlocked.Increment(ref index);

        ExtractorClass util = new ExtractorClass(SourceDbConnStr);
        util.LoadToFile(extractQuery, downLoadFileFullName);
        Interlocked.Increment(ref uiTimerIndex);
    });
});

Answer 1

我的猜测：

这看起来有很高的潜在 IO 来自：

• 数据库+磁盘
• 与 DB 和返回的网络通信
• 将结果写入磁盘

因此将花费大量时间等待 IO。我的猜测是，随着更多线程被添加到混合中并且 IO 被进一步强调，等待只会变得更糟。例如，一个磁盘只有一组磁头，因此您不能同时对其进行写入。如果您有大量线程尝试并发写入，则会降低性能。

尝试限制您正在使用的最大线程数：

var options = new ParallelOptions { MaxDegreeOfParallelism = 2 };

Parallel.ForEach(dtResult.AsEnumerable(), options, dr =>
{
    //Do stuff
});

更新

在您的代码编辑后，我建议您进行以下更改：

• 减少最大线程数 - 可以对此进行试验。
• 只执行一次目录检查和创建。

代码：

private static bool isDirectoryCreated;

//...

var options = new ParallelOptions { MaxDegreeOfParallelism = 2 };

Parallel.ForEach(dtResult.AsEnumerable(), options, dr =>
{
    string fileName, extractQuery, newDownLoadPath;

    lock (foreachObject)
    {
        fileName = extractorConfig.EncodeFileName(dr);

        extractQuery = extractorConfig.GetExtractQuery(dr);

        if (string.IsNullOrEmpty(extractQuery))
            throw new Exception("Extract Query not found. Please check the configuration");

        newDownLoadPath = CommonUtil.GetFormalizedDataPath(sDownLoadPath, uKey.CobDate);

        if (!isDirectoryCreated)
        {
            if (!Directory.Exists(newDownLoadPath))
                Directory.CreateDirectory(newDownLoadPath);

            isDirectoryCreated = true;
        }
    }

    string downLoadFileFullName = Path.Combine(newDownLoadPath, fileName);

    Interlocked.Increment(ref index);

    ExtractorClass util = new ExtractorClass(SourceDbConnStr);
    util.LoadToFile(extractQuery, downLoadFileFullName);

    Interlocked.Increment(ref uiTimerIndex);
});

Answer 2

如果没有相关代码，很难提供详细信息，但通常这是预期的行为。 .NET 尝试安排任务，使每个处理器都处于平均繁忙状态。

但这只能是近似的，因为并非所有任务都需要相同的时间。最后，一些处理器会完成工作，而另一些则不会，重新分配工作的成本很高，而且并不总是有益的。

我不知道 PLinq 使用的负载平衡的详细信息，但最重要的是，这种行为永远无法完全防止。

Answer 3

假设您将并行性限制为两个线程。 Parallel.ForEach 有（至少）两种可能的工作方式。一种方法是启动两个线程，每个线程都有一半的项目要完成。因此，如果您有 850 个项目，实际上会发生的是线程 1 被赋予前 425 个项目，线程 2 被赋予第二个 425 个项目块。现在两个线程都开始工作了。处理项目的顺序类似于：[0, 425, 426, 1, 2, 427, 3, 428, 429, 4, ...]。

很有可能（实际上很可能）其中一个线程比另一个线程更快地完成其项目组。

另一种可能的工作方式是启动两个线程，让每个线程从列表中获取一个项目，处理它，然后获取下一个项目，重复直到没有项目需要处理。在这种情况下，处理项目的顺序将更像 [0, 1, 2, 4, 3, 6, 5, ...]。

在第一个示例中，每个线程都有一个要处理的项目块。在第二种情况下，每个线程处理一个公共块中的项目，直到没有剩余的项目。

存在变体，但这是在多个线程之间分配工作的两种主要方式。要么给每个人自己的一组项目，要么期望每个线程在处理完一个项目后请求下一个项目。

Parallel.ForEach 以第一种方式实现：每个线程都有自己的一组要处理的项目。以其他方式执行此操作将需要更多开销，因为必须将项目列表视为共享队列，从而产生同步开销。