Node.js中Async/Await搭配createReadStream未读完就触发end事件问题

Node.js 原生的流事件回调不支持等待异步函数执行完成，你给data事件传入async回调时，逻辑执行规则如下：

• 当data事件触发，执行async回调碰到await关键字时，会立刻返回 pending 状态的 Promise，流会认为本次data事件的回调已经执行完成
• 你调用的stream.pause()只会阻止后续data事件触发，不会阻止end事件的触发：当流检测到所有数据都已经触发过data事件（不管回调里的异步逻辑有没有跑完），就会直接触发end事件
• 最后一行的校验逻辑是异步的，等await返回后才会执行推入数组和打印日志的操作，所以会出现在end事件的输出之后

推荐直接用 Node.js 支持的**异步迭代器for await...of**处理流，它会自动等待每次迭代的异步逻辑执行完成，所有数据处理完后才会执行后续逻辑，完全替代data和end事件的监听，代码更易维护：

const csvData = [];
try {
  // 创建流
  const stream = fs.createReadStream(filePath)
    .pipe(csv.parse({ headers: true }))
    .on("error", (error) => {
      throw error.message;
    });

  // 用for await迭代每一行
  for await (const row of stream) {
    const mobile = Helper.validateMobile(row.Telephone)
    const validationerrors = await Helper.ValidateCustomer(null, mobile, row.code, "BULK");
    console.log(`validationerrors ---> ${validationerrors} Telephone--> ${row.Telephone}`)
    if (validationerrors.length) {
      throw validationerrors.message
    } else {
      csvData.push(row);
    }
  }

  // 所有行处理完成后再执行原来end事件里的逻辑
  console.log("ENDDD ---->", csvData)
} catch (err) {
  // 统一处理错误
  throw err;
}

如果一定要用事件监听的方式实现，需要自行维护待处理的异步任务计数，等所有任务完成后再执行收尾逻辑：

const csvData = [];
let pendingTaskCount = 0;
let isStreamEnd = false;

const stream = fs.createReadStream(filePath)
  .pipe(csv.parse({ headers: true }))
  .on("error", (error) => {
    throw error.message;
  })
  .on("data", async (row) => {
    pendingTaskCount++;
    try {
      stream.pause();
      const mobile = Helper.validateMobile(row.Telephone)
      const validationerrors = await Helper.ValidateCustomer(null, mobile, row.code, "BULK");
      console.log(`validationerrors ---> ${validationerrors} Telephone--> ${row.Telephone}`)
      if (validationerrors.length) {
        throw validationerrors.message
      } else {
        csvData.push(row);
      }
    } finally {
      stream.resume();
      pendingTaskCount--;
      // 流已经触发end，且所有异步任务都处理完成，执行收尾逻辑
      if (isStreamEnd && pendingTaskCount === 0) {
        handleEnd()
      }
    }
  })
  .on("end", () => {
    isStreamEnd = true;
    // 确认没有待处理任务再执行收尾
    if (pendingTaskCount === 0) {
      handleEnd()
    }
  });

function handleEnd() {
  console.log("ENDDD ---->", csvData)
}

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