Stream Pipeline中途停止求助：200次迭代仅完成30次

使用Node.js stream/promises 的 pipeline 执行200条数据的处理流程时，仅完成30条就停止，无任何报错信息；但处理少量数据时运行正常。

相关代码 #

import { pipeline } from "stream/promises"

async function* func() {
  for (let i = 0; i < 200; i++) {
    console.log(`func ${i}`)
    yield i
  }
}
async function* func2(iterator: AsyncIterable<number>) {
  for await (let i of iterator) {
;    console.log(`func2 ${i}`)
    yield i
  }
}
async function* func3(iterator: AsyncIterable<number>) {
  for await (let i of iterator) {
    console.log(`func3 ${i}`)
    yield i
  }
}

async function main() {
  await pipeline(
    func,
    func2,
    func3
  )
}

main();

核心问题是Stream背压机制触发：Node.js Stream的背压逻辑会在下游无法消费数据时，暂停上游的数据生成。你的pipeline最后一步是func3这个异步生成器，但它产出的数据没有被任何下游消费者接收处理，当内部缓冲区被填满（默认容量对应约30条数据），背压会向上传递，最终导致func停止生成新数据，整个流程停滞。

另外注意func2代码里有一个多余的分号，虽然不影响流程逻辑，但建议清理以保证代码整洁。

需要在pipeline的最后添加一个数据消费步骤，确保func3产出的数据被完全消费，解除背压。以下是两种可行方案：

方案1：添加异步消费函数 #

import { pipeline } from "stream/promises"

async function* func() {
  for (let i = 0; i < 200; i++) {
    console.log(`func ${i}`)
    yield i
  }
}
async function* func2(iterator: AsyncIterable<number>) {
  for await (let i of iterator) {
    console.log(`func2 ${i}`)
    yield i
  }
}
async function* func3(iterator: AsyncIterable<number>) {
  for await (let i of iterator) {
    console.log(`func3 ${i}`)
    yield i
  }
}

// 新增消费函数，遍历处理所有剩余数据
async function consume(iterator: AsyncIterable<number>) {
  for await (const _ of iterator) {
    // 可在此添加实际业务逻辑，空遍历也可触发数据消费
  }
}

async function main() {
  await pipeline(
    func,
    func2,
    func3,
    consume // 加入消费步骤
  )
}

main();

方案2：使用可写流作为终点 #

import { pipeline } from "stream/promises"
import { Writable } from "stream"

// 生成器函数保持不变...

async function main() {
  await pipeline(
    func,
    func2,
    func3,
    Writable({
      objectMode: true,
      write(chunk, _, callback) {
        // 此处可添加数据处理逻辑
        callback() // 通知Stream可继续接收数据
      }
    })
  )
}

main();

两种方案都能确保200条数据被完整处理，不会因背压提前停止流程。

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