Node.js中如何用async/await异步使用createReadStream？CSV处理遇问题求助

嘿，这个问题我之前也碰到过，我来帮你捋清楚～

你遇到的核心问题是Node.js的流在触发data事件时不会等待异步操作完成。当fs.createReadStream开始读取文件后，只要CSV解析器把数据块解析成一行记录（row），就会立刻触发data事件——它完全不会等你上一个await processData(row)执行完毕。

所以实际过程是：流会快速把所有行的data事件都触发一遍，然后很快就读完文件触发end事件，但此时绝大多数processData的异步任务还在后台排队执行，这就导致end比所有处理完成早触发的现象。

这里有两种靠谱的解决思路，你可以根据需求选：

方法1：用pause/resume手动控制流节奏 #

这种方法很直接：每次触发data事件时先暂停流，等当前行的异步处理完成后再恢复流，确保一行处理完才会读下一行：

const fs = require('fs');
const { parse } = require('csv-parse');

async function processData(row) {
  // 你的异步操作，比如数据库写入、API调用等
  await someAsynchronousFunction();
}

const csvStream = fs.createReadStream('/file')
  .pipe(parse({ delimiter: ',', columns: true }));

csvStream.on('data', async (row) => {
  // 暂停流，避免继续触发data事件
  csvStream.pause();
  try {
    await processData(row);
  } catch (error) {
    console.error(`处理行出错:`, error);
  } finally {
    // 不管成功失败，都恢复流继续处理下一行
    csvStream.resume();
  }
});

csvStream.on('end', () => {
  console.log('done processing!');
});

方法2：用Transform流（更优雅的流模式） #

这种方法更贴合Node.js流的设计理念，通过创建一个Transform流插入到pipe链中，让流自动等待异步处理完成再继续：

const fs = require('fs');
const { parse } = require('csv-parse');
const { Transform } = require('stream');

async function processData(row) {
  await someAsynchronousFunction();
}

// 创建处理行的Transform流，开启objectMode因为csv解析输出的是对象
const rowProcessor = new Transform({
  objectMode: true,
  async transform(row, _, callback) {
    try {
      await processData(row);
      // 如果不需要把数据传给下游，第二个参数可以传null
      callback(null, row);
    } catch (error) {
      // 传递错误，会触发下游的error事件
      callback(error);
    }
  }
});

fs.createReadStream('/file')
  .pipe(parse({ delimiter: ',', columns: true }))
  .pipe(rowProcessor)
  .on('end', () => {
    console.log('done processing!');
  })
  .on('error', (error) => {
    console.error('流处理出错:', error);
  });

两种方法的对比 #

• 方法1适合简单场景，代码直观，容易理解；
• 方法2更适合复杂的流处理场景，比如后续还要加其他流操作（比如过滤、转换），错误处理也更统一，是Node.js推荐的流处理方式。

内容的提问来源于stack exchange，提问作者John Grayson