CsvHelper - 异步读取流

Question

我有一个服务，它接受一个包含需要批量插入数据库的 CSV 数据的输入流，并且我的应用程序尽可能使用 async/await。

过程是：使用CsvHelper的CsvParser解析流，将每一行添加到DataTable中，使用SqlBulkCopy将DataTable拷贝到数据库中。

数据可以是任何大小，所以我想避免一次将整个内容读入内存 - 显然我最终将在 DataTable 中拥有所有数据，因此在内存中基本上会有 2 个副本。

我想尽可能异步地完成所有这些，但 CsvHelper 没有任何异步方法，所以我想出了以下解决方法：

using (var inputStreamReader = new StreamReader(inputStream))
{
    while (!inputStreamReader.EndOfStream)
    {
        // Read line from the input stream
        string line = await inputStreamReader.ReadLineAsync();

        using (var memoryStream = new MemoryStream())
        using (var streamWriter = new StreamWriter(memoryStream))
        using (var memoryStreamReader = new StreamReader(memoryStream))
        using (var csvParser = new CsvParser(memoryStreamReader))
        {
            await streamWriter.WriteLineAsync(line);
            await streamWriter.FlushAsync();

            memoryStream.Position = 0;

            // Loop through all the rows (should only be one as we only read a single line...)
            while (true)
            {
                var row = csvParser.Read();

                // No more rows to process
                if (row == null)
                {
                    break;
                }

                // Add row to DataTable
            }
        }
    }
}

这个解决方案有什么问题吗？甚至有必要吗？我已经看到 CsvHelper 开发人员特别没有添加异步功能（https://github.com/JoshClose/CsvHelper/issues/202），但我并没有真正遵循不这样做的原因。

编辑：我刚刚意识到这个解决方案不适用于列包含换行符的情况:( 我猜我只需要将整个输入流复制到 MemoryStream 或其他东西

EDIT2：更多信息。

这是在一个库中的异步方法中，我试图在其中一直进行异步。它可能会被 MVC 控制器消耗（如果我只想从 UI 线程卸载它，我只会使用 Task.Run() 它）。大多数情况下，该方法将等待外部源，如数据库/DFS，我希望线程在它被释放时被释放。

CsvParser.Read() 即使阻塞的是读取流（例如，如果我试图读取的数据驻留在世界另一端的服务器上）也会阻塞，而如果 CsvHelper 要实现异步方法使用 TextReader.ReadAsync()，那么我就不会被阻止等待我的数据从迪拜到达。据我所知，我不是在要求围绕同步方法的异步包装器。

EDIT3：从遥远的未来更新！异步功能实际上是在 2017 年添加到 CsvHelper 中的。我希望我工作的公司的某个人从那时起就升级到了新版本！

Answer 1

Eric lippert 使用在餐厅做饭的比喻解释了 async-await 的用处。根据他的解释，如果您的线程没有其他事情可做，那么异步执行某些操作是没有用的。

另外，请注意，当您的线程正在执行某些操作时，它无法执行其他操作。只有当你的线程正在等待某件事时它才可以做其他事情。您在进程中等待的事情之一是读取文件。当线程逐行读取文件时，它必须等待几次才能读取行。在此等待期间，它可以执行其他操作，例如解析读取的 CSV 数据并将解析后的数据发送到目的地。

解析数据不是一个线程必须等待其他进程完成的过程，就像读取文件或将数据发送到数据库时必须做的那样。这就是解析过程没有异步版本的原因。普通的 async-await 不会让你的线程保持忙碌，因为在解析过程中没有什么可以等待的，所以在解析过程中你的线程没有时间做其他事情。

当然，您可以使用 Task.Run ( () => ParseReadData(...)) 将解析过程转换为可等待任务，并等待此任务完成，但以 Eric Lippert 的餐厅为例，这将是解冻一个做饭做这项工作，而你却坐在柜台后面无所事事。

但是，如果您的线程在解析读取的 CSV 数据时要做一些有意义的事情（例如响应用户输入），那么在单独的任务中开始解析可能会很有用。

如果你完整的读取-解析-更新数据库的过程不需要与用户交互，但是你需要你的线程在做这个过程的同时可以自由地做其他事情，可以考虑将完整的过程放在一个单独的任务中，并启动该任务无需等待。在这种情况下，您仅使用界面线程来启动其他任务，并且您的界面线程可以自由地执行其他操作。与整个过程的总时间相比，开始这项新任务的成本相对较小。

再说一遍：如果你的线程没有别的事可做，就让这个线程来做处理，不要启动其他任务来做它。