如何使用Parallel.ForEach正确写入文件？

Question

我有一个任务,它逐行读取一个大文件,用它做一些逻辑,并返回一个我需要写入文件的字符串.输出的顺序无关紧要.但是,当我尝试下面的代码时,它会在读取我的文件的15-20k行后停止/变得非常慢.

public static Object FileLock = new Object();
...
Parallel.ForEach(System.IO.File.ReadLines(inputFile), (line, _, lineNumber) =>
{
    var output = MyComplexMethodReturnsAString(line);
    lock (FileLock)
    {
        using (var file = System.IO.File.AppendText(outputFile))
        {
            file.WriteLine(output);
        }
    }
});

为什么我的程序在运行一段时间后会变慢？有没有更正确的方法来执行此任务？

Answer 1

您基本上通过让所有线程都尝试写入文件来序列化您的查询.相反,您应该计算需要写入的内容,然后在结束时编写它们.

var processedLines = File.ReadLines(inputFile).AsParallel()
    .Select(l => MyComplexMethodReturnsAString(l));
File.AppendAllLines(outputFile, processedLines);

如果您需要在数据到来时刷新数据,请打开流并启用自动刷新(或手动刷新):

var processedLines = File.ReadLines(inputFile).AsParallel()
    .Select(l => MyComplexMethodReturnsAString(l));
using (var output = File.AppendText(outputFile))
{
    output.AutoFlush = true;
    foreach (var processedLine in processedLines)
        output.WriteLine(processedLine);
}

Answer 2

这与Parallel.ForEach内部负载均衡器的工作原理有关.当它发现您的线程花费大量时间阻塞时,它会通过在问题上抛出更多线程来加快速度,从而导致更高的并行开销,争用FileLock和整体性能下降.

为什么会这样？因为Parallel.ForEach不适合IO工作.

你怎么解决这个问题？使用Parallel.ForEach仅适用于CPU工作和执行并行循环之外的所有IO.

一个快速的解决方法是Parallel.ForEach通过使用接受的重载来限制允许登记的线程数ParallelOptions,如下所示:

Parallel.ForEach(
    System.IO.File.ReadLines(inputFile),
    new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
    (line, _, lineNumber) =>
    {
        ...
    }