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

问题描述

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

I am having difficulty with using fs.creadReadStream to process my csv file asynchronously:

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')之前，逐一读取每条记录，然后执行一些异步功能.

I want to perform some asynchronous function after reading each record one by one before the createReadStream reaches on('end').

但是，在我所有数据完成处理之前， on('end')被命中了.有人知道我可能在做错什么吗?

However, the on('end') gets hit before all of my data finishes processing. Does anyone know what I might be doing wrong?

提前谢谢！

推荐答案

.on('data，...)不等待您的 await .记住，一个 async 函数会立即返回一个诺言，而 .on()对该诺言没有给予任何关注，因此它会保持愉快的状态.

.on('data, ...) does not wait for your await. Remember, an async function returns a promise immediately and .on() is not paying any attention to that promise so it just keeps merrily going on.

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

The await only waits inside the function, it does not stop your function from returning immediately and thus the stream thinks you've process the data and keeps sending more data and generating more data events.

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

There are several possible approaches here, but the simplest might be to pause the stream until processData() is done and then restart the stream.

此外， processData()是否返回与异步操作完成相关的promise?这也是 await 能够完成其工作所必需的.

Also, does processData() return a promise that is linked to the completion of the async operation? That is also required for await to be able to do its job.

可读流文档包含在期间暂停流的示例数据事件，然后在一些异步操作完成后恢复它.这是他们的示例:

The readable stream doc contains an example of pausing the stream during a data event and then resuming it after some asynchronous operation finishes. Here's their example:

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);
});

这篇关于如何使用async/await在node.js中异步创建ReadStream的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！