Node.js csv-parse流式处理CSV时await未完成就触发end的解决方法

csv-parse的事件回调默认不等待异步操作执行：data事件触发时，传入的async函数会返回Promise，但事件发射器本身不会追踪这个Promise的状态，只要文件内容全部读完就会立刻触发end事件，此时队列里的数据库写入异步操作大概率还没执行完，就会出现提前触发的问题。
如果不对写入节奏做控制，大文件场景下短时间内会堆积大量数据库请求，很容易打满数据库连接池，甚至拖垮服务。

针对大文件流式处理、收集全量写入错误的需求，最稳妥的方式是使用csv-parse原生支持的异步迭代器API，不需要额外引入第三方流控制依赖，天然适配async/await逻辑，能保证所有行的异步操作执行完成后再走到收尾逻辑。
参考实现代码如下：

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

/**
 * 流式处理CSV文件，收集存入数据库的错误
 * @param {string} csvFilePath CSV文件路径
 * @returns {Promise<Array>} 写入错误集合
 */
async function processCsvFile(csvFilePath) {
  const saveErrors = [];
  // 初始化文件读取流 + CSV解析流，全程不把全量文件加载到内存
  const csvParser = fs.createReadStream(csvFilePath).pipe(
    parse({
      columns: true, // 自动识别表头，返回key-value格式的行对象，可根据需求调整
      skip_empty_lines: true, // 跳过空行
      // 其余解析配置可根据业务需求添加
    })
  );

  // 逐行迭代处理，每一行的数据库操作完成后才会处理下一行
  for await (const [lineNum, row] of csvParser.iterator()) {
    try {
      await tryToSaveToDatabase(row);
    } catch (err) {
      // 记录错误时带上行号，方便前端定位问题行
      saveErrors.push({
        lineNumber: lineNum,
        rowContent: row,
        errorMessage: err.message
      });
    }
  }

  // 执行到此处时所有行处理、所有数据库异步操作均已完成
  return saveErrors;
}

// 调用示例
processCsvFile('./target.csv')
  .then(errors => {
    // 此处拿到全量错误，可直接返回给前端
    console.log(`CSV处理完成，共发现${errors.length}条写入错误`);
  })
  .catch(fatalErr => {
    // 捕获文件不存在、CSV格式非法等全局致命错误
    console.error('CSV处理流程异常', fatalErr);
  });

• 单条写入数据库性能较差时，可以攒小批量数据（比如50~200条）做批量插入，注意控制批次大小，避免内存占用过高
• 单条写入耗时较长时，可以加简单的并发控制（比如同时最多执行3~5个写入任务），在不打垮数据库的前提下提升处理速度，不要无限制并发
• 错误信息必须携带行号或者业务唯一标识，否则前端拿到错误也无法对应到具体的问题行
• 禁止先把整个CSV文件读取为字符串再传入解析器，必须用文件流对接解析流，否则GB级大文件会直接占满服务内存

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