尝试有效地并行处理 .NET 中文件流中的一些数据（读写）

Question

我正在尝试了解如何有效地从文件中读取一些数据，执行一些并行工作（每行），然后将新行写回文件系统。

我知道我可以做到这一点，一次一行.. 但我希望一次做几行 - 或者 - .. 如果一行“忙”等待异步工作完成，然后继续下一行，等等。

这里有一些示例数据和逻辑...

Header
SomeId#1, SomeId#2, SomeId#3, Name, Has this line been processed and cleaned(true/false)

File Data
444,2,12,Leia Organa, true
121,33333,4,Han Solo, true
1,2,3,Jane Doe, false
1,4,11,John Doe, false

所以前两行已经处理完毕，我将跳过这些行。 第 3 行和第 4 行需要处理。检查数据后，我希望将其保存回文件中，如

1,33333,3,Jane Doe, true

所以这是一般逻辑...

• 读线
• 致电DoWorkAsync()
• 再次将此行保存回文件。

我只是希望不必等待DoWorkAsync() 完成，然后我就可以保存然后阅读下一行。我希望我可以开始阅读下一行......如果上一行完成......很好......然后将该行保存到文件中的相同行号......然后再次移动到下一行。

这就像我可以同时工作 5 或 10 行......等待结果从 3rd 方 api 返回......并行工作或其他。

这可以在 .NET 中完成吗？我确定 .NET 具有此功能.. 我只是看不到执行此操作的模式。

注意：我通常对 I/O 密集型操作（例如访问文件系统或调用某些 3rd 方 api 端点）执行 async/await 与我用于 CPU 密集型工作的 Parallel.ForEach。

注意：为什么true/false 在行尾？因为我不能一次处理所有的行。我有 api 限制。

其他想法是有两个文件，一个用于 PENDING，一个用于 PROCESSED。

Answer 1

这是一个并行处理器的存根，它在批量处理行时使用async/await。

这种方法确保写入时保持相同的顺序。

public async Task ProcessFile()
{
    const int parallelism = 5;

    using (var readStream = File.OpenRead(@"c:\myinputfile"))
    {
        // put HERE your logic for skipping to a specific line
        // e.g. readStream.Seek(lastPosition); 

        using (var reader = new StreamReader(readStream))
        {
            while (!reader.EndOfStream)
            {
                var tasks = new List<Task<string>>();

                for (var i = 0; i < parallelism; i++)
                {   
                    var line = await reader.ReadLineAsync();

                    tasks.Add(DoWorkAsync(line));

                    if (reader.EndOfStream)
                        break;
                }

                var results = await Task.WhenAll(tasks);

                using (var writeStream = File.Open(@"d:\myresultfile", FileMode.Append))
                using (var writer = new StreamWriter(writeStream))
                {
                    foreach (var line in results)
                        await writer.WriteLineAsync(line);
                }
            }
        }
    }
}

public async Task<string> DoWorkAsync(string line)
{
    await Task.Delay(new Random().Next(1000, 5000));
    // do some work and return line with last parameter = true
    return line.Replace("false", "true"); // e.g.
}

它肯定需要改进，但它应该为你自己编写一个良好的基础。