嵌套for循环中使用Node.js createWriteStream写入海量数据文件为空问题

问题根源 #

• 未处理流的背压机制：fs.WriteStream.write() 方法会返回布尔值，返回false时表示内部缓冲区已超出highWaterMark阈值，此时继续同步调用write会导致数据堆积在内存中，无法落盘，进程结束时缓冲区数据未刷入磁盘就被销毁，最终输出文件为空。
• 冗余的流关闭操作：调用fileout.end()后无需再执行fileout.close()，end()会自动在所有待写数据刷入磁盘后关闭流，额外的close()调用可能会中断未完成的写入流程。
• process.stdout也存在背压风险：如果进程标准输出被重定向到文件，频繁的stdout.write()也会触发背压，阻塞整个写入流程。

修复后的实现代码 #

const fs = require('fs');
const { once } = require('events');

const words = fs.readFileSync('./1024words.txt').toString().split("\n").filter(Boolean); // 过滤空行
const totRows = Math.pow(words.length, 3);
let count = 0;

// 异步写入工具函数，处理背压
async function writeRow(stream, row) {
  const canWrite = stream.write(row);
  if (!canWrite) {
    // 缓冲区满，等待drain事件触发后再继续写入
    await once(stream, 'drain');
  }
}

(async () => {
  for (let indexA = 0; indexA < words.length; indexA++) {
    const wordA = words[indexA];
    // 增大缓冲区到16MB，减少drain触发频次，提升写入性能
    const fileout = fs.createWriteStream(`${wordA}.csv`, { highWaterMark: 1024 * 1024 * 16 });
    
    for (let indexB = 0; indexB < words.length; indexB++) {
      const wordB = words[indexB];
      for (let indexC = 0; indexC < words.length; indexC++) {
        const wordC = words[indexC];
        const row = `${wordA} ${wordB} ${wordC}\n`;
        count++;
        // 每10万行输出一次进度，减少stdout压力
        if (count % 100000 === 0) {
          console.log(`${count} / ${totRows} 已写入`);
        }
        await writeRow(fileout, row);
      }
    }
    
    // 结束写入，自动关闭流
    fileout.end();
    // 等待流完全关闭再处理下一个文件，避免资源占用过高
    await once(fileout, 'close');
  }
  console.log('所有文件写入完成');
})();

优化建议 #

• 可根据服务器内存情况调整highWaterMark的数值，内存充足的情况下设置更大的缓冲区可以进一步提升写入速度。
• 如果不需要实时进度，可以直接关闭进度输出，写入性能会有更明显的提升。

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