如何在async/await下向NodeCSV stringify流写入时避免背压？

问题背景 #

用node-csv工具将对象数组写入CSV文件时，直接调用await csv.write(row)会触发超时错误：

Error: ETIMEDOUT: connection timed out, write
at writeSync (node:fs:933:3)

但在每次写入后加await timeout(0)就能解决问题，推测是背压超出容量导致——原本以为await write会自动等待流准备就绪，实际并非如此。

原代码如下：

async function(outstream) { 
     const csv = CSV.stringify({header: true})
     csv.pipe(outstream)
     for (let row of rows) {
        await csv.write(row);
        // await timeout(0)  
     }
}

原因分析 #

node-csv的csv.write(row)返回的Promise并没有正确处理流的背压状态：当输出流（比如网络流、慢磁盘流）处理速度跟不上写入速度时，流会暂时暂停接收数据，此时继续写入会导致数据在缓冲中堆积，最终触发超时。而await timeout(0)是把当前任务放到事件循环末尾，给了流处理缓冲、恢复可写状态的时间，相当于手动让出事件循环，临时缓解了背压，但这是个hack方案。

正确解决方法 #

方法一：利用流原生的drain事件处理背压 #

通过监听drain事件，确保只有当流准备好接收更多数据时，才写入下一行：

async function writeToCsv(outstream, rows) {
  const csv = CSV.stringify({header: true})
  csv.pipe(outstream)

  for (const row of rows) {
    // 尝试写入，返回值表示是否还能继续写入
    const canContinueWriting = csv.write(row)
    if (!canContinueWriting) {
      // 等待drain事件，说明流缓冲已清空，可继续写入
      await new Promise(resolve => csv.once('drain', resolve))
    }
  }
  // 必须调用end()，确保所有缓冲数据都被写入输出流
  csv.end()
}

方法二：用可读流+管道自动处理背压 #

把对象数组转成可读流，通过管道连接到csv转换器和输出流，Node.js会自动处理背压，无需手动控制写入节奏：

const { Readable } = require('stream')

async function writeToCsv(outstream, rows) {
  const csv = CSV.stringify({header: true})
  // 将对象数组转为可读流（objectMode设为true支持对象输入）
  const rowStream = Readable.from(rows, { objectMode: true })
  
  // 管道链自动处理背压
  rowStream.pipe(csv).pipe(outstream)

  // 等待所有数据处理完成
  await new Promise((resolve, reject) => {
    outstream.on('finish', resolve)
    outstream.on('error', reject)
  })
}

关键说明 #

• 不要依赖timeout(0)这种hack方案，它只是临时绕过了问题，并没有从根本上处理背压。
• Node.js流的drain事件是处理背压的标准方式，所有符合规范的流都会触发这个事件。
• 用管道（pipe）的方式是最省心的，Node.js会自动协调上下游的速度，避免缓冲堆积。

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