如何使用 TPL 并行化文件写入？

Question

我正在尝试将字符串列表保存到多个文件中，每个字符串在不同的文件中，并同时进行。我是这样做的：

public async Task SaveToFilesAsync(string path, List<string> list, CancellationToken ct)
{
    int count = 0;
    foreach (var str in list)
    {
        string fullPath = path + @"\" + count.ToString() + "_element.txt";
        using (var sw = File.CreateText(fullPath))
        {
            await sw.WriteLineAsync(str);
        }
        count++;

        NLog.Trace("Saved in thread: {0} to {1}", 
           Environment.CurrentManagedThreadId,
           fullPath);

        if (ct.IsCancellationRequested)
            ct.ThrowIfCancellationRequested();
    }
}

然后这样称呼它：

try
{
   var savingToFilesTask = SaveToFilesAsync(@"D:\Test", myListOfString, ct);
}
catch(OperationCanceledException)
{
   NLog.Info("Operation has been cancelled by user.");
}

但是在日志文件中我可以清楚地看到保存总是发生在同一个线程 id 中，所以没有并行性？我究竟做错了什么？如何解决？我的目标是使用所有计算机内核尽可能快地节省所有费用。

Answer 1

本质上，您的问题是foreach 是同步的。它使用同步的IEnumerable。

要解决这个问题，首先将循环体封装到一个异步函数中。

public async Task WriteToFile(
        string path,
        string str,
        int count)
{
    var fullPath = string.Format("{0}\\{1}_element.txt", path, count);
    using (var sw = File.CreateText(fullPath))
    {
        await sw.WriteLineAsync(str);
    }

    NLog.Trace("Saved in TaskID: {0} to \"{1}\"", 
       Task.CurrentId,
       fullPath);
}

然后，不是同步循环，而是将字符串序列投射到执行封装循环体的任务序列中。这本身不是异步操作，但投影不会阻塞，即没有await。

然后等待他们所有的任务按照任务调度器定义的顺序完成。

public async Task SaveToFilesAsync(
        string path,
        IEnumerable<string> list,
        CancellationToken ct)
{
    await Task.WhenAll(list.Select((str, count) => WriteToFile(path, str, count));
}

没有什么要取消的，所以没有必要向下传递取消令牌。

我使用了Select 的索引重载来提供count 值。

我已将您的日志记录代码更改为使用当前的任务 ID，这避免了调度方面的任何混乱。

Answer 2

如果您想进行并行处理，您必须告诉 .NET 这样做。 我认为，如果您将代码拆分为一个附加函数，那么最简单的方法之一就会变得清晰。

这个想法是将实际的单个 IO 操作拆分为一个额外的异步函数并调用这些函数而不等待它们，而是将它们存储在一个列表中并在最后等待所有它们。

我通常不写 C# 代码，所以请原谅我可能犯的任何语法错误：

public async Task SaveToFilesAsync(string path, List<string> list, CancellationToken ct)
{
    int count = 0;
    var writeOperations = new List<Task>(list.Count);
    foreach (var str in list)
    { 
        string fullPath = path + @"\" + count.ToString() + "_element.txt";
        writeOperations.add(SaveToFileAsync(fullPath, str, ct));
        count++;
        ct.ThrowIfCancellationRequested();
    }

    await Task.WhenAll(writeOperations);
}

private async Task SaveToFileAsync(string path, string line, CancellationToken ct)
{
    using (var sw = File.CreateText(path))
    {
        await sw.WriteLineAsync(line);
    }

    NLog.Trace("Saved in thread: {0} to {1}", 
        Environment.CurrentManagedThreadId, 
        fullPath);

    ct.ThrowIfCancellationRequested();
}

这样 IO 操作由同一个线程一次又一次地触发。这应该工作得非常快。一旦使用 .NET ThreadPool 完成 IO 操作，就会触发延续。

我还删除了if (ct.IsCancellationRequested) 检查，因为无论如何这是由ct.ThrowIfCancellationRequested(); 完成的。

希望这能让您了解如何处理这些事情。

Answer 3

如果这是在并行存储 (SSD) 上，您可以通过并行化来加快速度。由于没有内置方法可以以一定的并行度并行化异步循环，因此我建议使用具有固定并行度和同步 IO 的 PLINQ。 Parallel.ForEach 不能有一个固定的 DOP（只有一个最大 DOP）。

Answer 4

我已经在原始问题中添加了我的答案，我应该在此处添加吗？ C# TPL calling tasks in a parallel manner and asynchronously creating new files

编辑：这是现在并行运行多个保存的建议解决方案。

您需要将 foreach 循环（从第一项到最后一项按顺序运行）替换为可配置为并行性的 Parallel.ForEach() 循环。

var cts = new CancellationTokenSource();
Task.WaitAll(SaveFilesAsync(@"C:\Some\Path", files, cts.Token));
cts.Dispose();

然后在该方法中进行并行处理。

public async Task SaveFilesAsync(string path, List<string> list, CancellationToken token)
{
    int counter = 0;

    var options = new ParallelOptions
                      {
                          CancellationToken = token,
                          MaxDegreeOfParallelism = Environment.ProcessorCount,
                          TaskScheduler = TaskScheduler.Default
                      };

    await Task.Run(
        () =>
            {
                try
                {
                    Parallel.ForEach(
                        list,
                        options,
                        (item, state) =>
                            {
                                // if cancellation is requested, this will throw an OperationCanceledException caught outside the Parallel loop
                                options.CancellationToken.ThrowIfCancellationRequested();

                                // safely increment and get your next file number
                                int index = Interlocked.Increment(ref counter);
                                string fullPath = string.Format(@"{0}\{1}_element.txt", path, index);

                                using (var sw = File.CreateText(fullPath))
                                {
                                    sw.WriteLine(item);
                                }

                                Debug.Print(
                                    "Saved in thread: {0} to {1}",
                                    Thread.CurrentThread.ManagedThreadId,
                                    fullPath);
                            });
                }
                catch (OperationCanceledException)
                {
                    Debug.Print("Operation Canceled");
                }
            });
}