Node.js异步迭代器处理大CSV：两种批量处理方案优劣分析

需求目标 #

实现可处理超大文件的快速、可扩展解决方案，核心流程：

1. 逐行读取CSV文件
2. 将每行数据输入复杂函数，生成对应单个文件（一行对应一个文件）
3. 所有文件生成完成后，压缩为ZIP包

现有实现分析 #

步骤1：逐行读取CSV文件 #

基于CSV Parser的async iterator API实现，流式读取避免加载整个文件到内存，天然适配超大文件场景：

async function* loadCsvFile(filepath, params = {}) {
  try {
    const parameters = {
      ...csvParametersDefault,
      ...params,
    };
    const inputStream = fs.createReadStream(filepath);
    const csvParser = parse(parameters);
    const parser = inputStream.pipe(csvParser)
    for await (const line of parser) {
      yield line;
    }
  } catch (err) {
    throw new Error("error while reading csv file: " + err.message);
  }
}

步骤2：行数据处理的两种方案对比 #

方案1：串行处理

逐个await耗时操作handleCsvLine：

// step 1
const csvIterator = loadCsvFile(filePath, options);
// step 2
let counter = 0;
for await (const row of csvIterator) {
  await handleCvsLine(row);
  counter++;
  if (counter % 50 === 0) {
    logger.debug(`Processed label ${counter}`);
  }
}
// step 3
zipFolder(folderPath);

优点：

• 可直观看到文件逐个生成
• 进度展示准确，每完成一行就更新计数

缺点：

• 串行执行导致CPU、磁盘IO资源利用率极低，处理速度极慢，完全不适合超大文件场景

方案2：无限制并发处理

将所有handleCsvLine任务存入数组，最后用Promise.all批量执行：

// step 1
const csvIterator = loadCsvFile(filePath, options);
// step 2
let counter = 0;
const promises = [];
for await (const row of csvIterator) {
  promises.push(handleCvsLine(row));
  counter++;
  if (counter % 50 === 0) {
    logger.debug(`Processed label ${counter}`);
  }
}

await Promise.all(promises);
// step 3
zipFolder(folderPath);

优点：

• 并行执行任务，理论上速度远快于串行

缺点：

• 对百万行级别的超大文件，会瞬间创建大量Promise和文件操作，导致内存暴涨、文件句柄耗尽，触发系统报错
• 进度展示失真：循环很快读完所有行，但实际任务还在后台执行，后续长时间无进度更新，用户体验极差

最优实现方案：带并发控制的批量处理 #

结合前两种方案的优点，通过限制并发任务数量，既保证处理速度，又避免资源耗尽，同时提供准确的进度反馈。

实现代码 #

// 步骤1：读取CSV
const csvIterator = loadCsvFile(filePath, options);
// 步骤2：带并发控制的处理
const concurrencyLimit = 10; // 根据系统资源调整（如CPU核心数、磁盘IO能力）
let activePromises = [];
let totalRead = 0;
let totalCompleted = 0;

for await (const row of csvIterator) {
  // 创建任务Promise，完成后更新完成计数
  const taskPromise = handleCvsLine(row)
    .then(() => {
      totalCompleted++;
      if (totalCompleted % 50 === 0) {
        logger.debug(`Completed ${totalCompleted} items (total read: ${totalRead})`);
      }
    })
    .catch(err => {
      logger.error(`Failed to process item ${totalRead + 1}: ${err.message}`);
    });

  activePromises.push(taskPromise);
  totalRead++;

  // 达到并发限制时，等待至少一个任务完成，释放资源
  if (activePromises.length >= concurrencyLimit) {
    await Promise.race(activePromises);
    // 过滤掉已完成的Promise，保留未完成的任务
    activePromises = activePromises.filter(p => p.status !== 'fulfilled' && p.status !== 'rejected');
  }
}

// 等待剩余所有任务完成
await Promise.all(activePromises);
// 步骤3：压缩文件夹
zipFolder(folderPath);

方案优势 #

1. 资源可控：限制并发数，避免内存过载和文件句柄耗尽，稳定支持超大文件处理
2. 效率较高：并行执行任务，资源利用率远高于串行，速度接近无限制并发的性能
3. 进度准确：同时统计已读取行数和已完成行数，用户能清晰看到处理进度
4. 容错性强：单个任务失败不影响整体流程，可记录错误信息便于排查

步骤3：文件夹压缩实现 #

使用archiver库实现压缩，注意处理警告和错误：

function zipFolder(folderPath, globPath, outputFolder, outputName, logger) {
  return new Promise((resolve, reject) => {
    // 创建压缩文件输出流
    const stream = fs.createWriteStream(path.join(outputFolder, outputName));
    const archive = archiver("zip", {
      zlib: { level: 9 }, // 最高压缩率，若追求速度可改为5-6
    });

    archive.glob(globPath, { cwd: folderPath });

    // 处理警告（如文件不存在）
    archive.on("warning", function (err) {
      if (err.code === "ENOENT") {
        logger.warning(err);
      } else {
        logger.error(err);
        reject(err);
      }
    });

    // 处理压缩错误
    archive.on("error", function (err) {
      logger.error(err);
      reject(err);
    });

    // 管道输出压缩数据
    archive.pipe(stream);
    // 监听压缩完成事件
    stream.on("close", function () {
      resolve();
    });
    archive.finalize();
  });
}

压缩优化建议 #

• 若追求处理速度，可降低zlib.level到5-6，平衡压缩率和速度
• 若生成的文件数量极大，可在处理行数据时记录文件路径，直接调用archive.file()添加文件，避免使用glob遍历目录，提升压缩效率

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