如何使用 async/await 在 node.js 中异步创建ReadStream答案

【问题标题】：How to asynchronously createReadStream in node.js with async/await如何使用 async/await 在 node.js 中异步创建ReadStream
【发布时间】：2018-05-12 14:11:49
【问题描述】：

我无法使用 fs.creadReadStream 异步处理我的 csv 文件：

async function processData(row) {
    // perform some asynchronous function
    await someAsynchronousFunction();
}

fs.createReadStream('/file')
    .pipe(parse({
        delimiter: ',',
        columns: true
    })).on('data', async (row) => {
        await processData(row);
    }).on('end', () => {
        console.log('done processing!')
    })

我想在createReadStream到达on('end')之前一一读取每条记录后执行一些异步功能。

但是，on('end') 在我的所有数据完成处理之前就被命中了。有谁知道我可能做错了什么？

提前致谢！

【问题讨论】：

标签： node.js stream async-await

【解决方案1】：

.on('data, ...) 不会等待您的await。请记住，async 函数会立即返回一个 Promise，而 .on() 并没有关注该 Promise，所以它只是继续愉快地进行。

await 仅在函数内部等待，它不会阻止您的函数立即返回，因此流认为您已处理数据并继续发送更多数据并生成更多 data 事件。

这里有几种可能的方法，但最简单的可能是暂停流直到processData() 完成，然后重新启动流。

另外，processData() 是否返回与异步操作完成相关的承诺？这也是await 能够完成其工作所必需的。

readable stream doc 包含在 data 事件期间暂停流，然后在某些异步操作完成后恢复流的示例。这是他们的例子：

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  readable.pause();
  console.log('There will be no additional data for 1 second.');
  setTimeout(() => {
    console.log('Now data will start flowing again.');
    readable.resume();
  }, 1000);
});

【讨论】：

【解决方案2】：

我最近遇到了同样的问题。我通过使用一系列 promise 来修复它，并在触发 .on("end") 时等待它们全部解决。

import parse from "csv-parse";

export const parseCsv = () =>
  new Promise((resolve, reject) => {
    const promises = [];
    fs.createReadStream('/file')
      .pipe(parse({ delimiter: ',', columns: true }))
      .on("data", row => promises.push(processData(row)))
      .on("error", reject)
      .on("end", async () => {
        await Promise.all(promises);
        resolve();
      });
  });

【讨论】：

这就是我要找的。非常感谢。 @TFischer