Node.js 的 stream/promises.pipeline 在链式异步迭代器中若末尾函数仍 yield 数据但无消费者，会导致背压累积、流挂起，表现为处理中途静默停止（如 200 条仅处理前 30 条）。根本原因在于缺少最终消费端，数据滞留在内部缓冲区。
node.js 的 `stream/promises.pipeline` 在链式异步迭代器中若末尾函数仍 `yield` 数据但无消费者，会导致背压累积、流挂起，表现为处理中途静默停止（如 200 条仅处理前 30 条）。根本原因在于缺少最终消费端，数据滞留在内部缓冲区。
在使用 pipeline 组合多个 async generator function（即 AsyncIterable）时，需严格遵循“生产者–转换器–消费者”语义。pipeline(a, b, c) 内部会将 a 作为源（source），b 作为中间转换流（transform），而 最后一个参数 c 默认被视为目标（destination） —— 它应当消费数据，而非继续产出。
你遇到的问题（200 次迭代仅执行约 30 次后静默终止）正是典型的背压未释放现象：
-
func生成数字 0~199； -
func2接收并原样yield转发； -
func3同样接收并yield—— 但此时已无下游接收者，所有yield值被暂存在pipeline内部缓冲区中； - 当缓冲区填满（默认
highWaterMark = 32，实际触发点常在 ~30 左右），上游for await循环因await内部next()调用被阻塞，整个流暂停，且无错误抛出（符合 Node.js 流的背压设计）。
✅ 正确做法是：确保 pipeline 的最后一个函数不 yield，而是完成实际消费逻辑（如写入文件、存入数据库、聚合统计等）。例如：
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 (const i of iterator) {
console.log(`func2 ${i}`);
yield i; // ✅ 中间转换器可 yield
}
}
// ✅ 最终 destination：只消费，不 yield
async function func3(iterator: AsyncIterable<number>) {
let count = 0;
for await (const i of iterator) {
console.log(`func3 ${i}`);
count++;
}
console.log(`✅ 全部 ${count} 项处理完成`);
}
async function main() {
await pipeline(func, func2, func3); // 注意：func3 现为普通 async function（非 async*）
}
main().catch(console.error);⚠️ 关键注意事项：
-
pipeline的 destination 参数可以是AsyncIterable（此时需确保其内部不产出数据），但更推荐使用普通async function作为终点，语义清晰且不易误用； - 若需在终点做异步操作（如
fetch、writeFile），务必await它们，避免未处理 Promise 导致流程提前结束； - 不要依赖
console.log判断流是否完成——它可能因缓冲延迟而滞后；应以终点函数的return或显式console.log('done')为准； - 如需调试背压行为，可显式设置
highWaterMark（如{ highWaterMark: 1 }加速复现问题），或监听'drain'事件（仅适用于传统Writable流，async generatordestination 不直接暴露该事件）。
总结：pipeline 的健壮性高度依赖各阶段的角色一致性。记住口诀——“源产生，中间转换，终点消费；终点一旦 yield，流必卡在半路。”
