stream.pipeline 比手写 .pipe() 链更可靠，因其内置统一错误传播、自动销毁未完成流及反向信号中止；手动链易漏错误监听和资源清理，导致 err_stream_premature_close 或进程卡死。
stream.pipeline 为什么比手写 .pipe() 链更可靠
处理 GB 级文件时，手写 a.pipe(b).pipe(c) 极易漏掉错误监听和资源清理——比如某个中间流提前结束、背压未正确传递、或异常发生后后续流仍在运行。而 stream.pipeline 内置了统一的错误传播、自动销毁未完成流、以及反向信号中止（如下游拒绝接收，上游立刻暂停），这是它成为大文件流式处理事实标准的关键原因。
常见错误现象：ERR_STREAM_PREMATURE_CLOSE 或进程卡住不退出，往往就是手动链中某处没监听 error 或没调用 destroy()；stream.pipeline 默认帮你兜底。
- 必须传入至少两个流，否则抛
TypeError: streams[0] is not a stream - 所有流必须是可读、可写或双工流；不能混入 Promise 或普通函数
- 不支持 Node.js
如何用 pipeline 实现边读边转码再上传（例如 CSV → JSONL → S3）
典型场景：读取一个 5GB 的原始 CSV 文件，逐行解析为对象，序列化成 JSONL，再通过 @aws-sdk/client-s3 的 PutObjectCommand 流式上传——整个过程内存占用应稳定在几十 MB，而非加载全量数据。
核心在于每一步都使用真正支持背压的流：用 fs.createReadStream（配 highWaterMark: 64 * 1024）、csv-parse 的 parse 流（设 {objectMode: true}）、JSONStream.stringify 或自定义 Transform 流、最后接 S3 的 Upload 流（需包装成 Writable）。
示例关键片段：
const { pipeline } = require('stream');
const fs = require('fs');
const parse = require('csv-parse/lib/sync'); // 注意：这里用的是 sync 版本，实际应换为流式 parse()
// 正确做法是用 csv-parse 的流式 parser：
const { parse } = require('csv-parse');
const { Transform } = require('stream');
<p>const csvParser = parse({ columns: true, skipEmptyLines: true });
const jsonlTransformer = new Transform({
objectMode: true,
transform(chunk, encoding, callback) {
try {
this.push(JSON.stringify(chunk) + '\n');
callback();
} catch (err) {
callback(err);
}
}
});</p><p>pipeline(
fs.createReadStream('huge.csv', { highWaterMark: 64 * 1024 }),
csvParser,
jsonlTransformer,
s3UploadWritable, // 自定义 Writable 流，内部调用 S3.upload()
(err) => {
if (err) console.error('Pipeline failed:', err);
else console.log('Upload done');
}
);</p>
highWaterMark 设置不当会导致 OOM 或吞吐骤降
highWaterMark 不是越大越好。对 GB 文件，设成 100 * 1024 * 1024（100MB）看似能减少 I/O 次数，实则让内存堆积大量未消费数据，一旦下游处理慢（比如网络上传延迟），Node.js 事件循环被阻塞，V8 堆内存暴涨，最终触发 FATAL ERROR: Ineffective mark-compacts near heap limit。
- 读取文件：推荐
64 * 1024到256 * 1024（64–256KB），平衡系统调用开销与内存驻留 - 解析/转换流：若处理逻辑重（如正则匹配、JSON 解析），应同步降低
highWaterMark，甚至设为1（逐个对象处理） - 写入远程服务（如 S3）：务必检查 SDK 是否真正支持流式写入；
@aws-sdk/client-s3v3 的PutObjectCommand接收Readable，但底层仍会缓冲分块——此时应配合upload的partSize参数（默认 5MB），避免单次上传块过大
错误恢复与部分失败时的 cleanup 容易被忽略
stream.pipeline 虽自动调用 destroy()，但它不会回滚已写入的部分数据。比如上传到 S3 时中途失败，可能已有前几个 MB 被写入，但 S3 不提供“原子截断”操作——你得自己记录 checkpoint 或改用分段上传（CreateMultipartUpload）并手动 abort。
另一个常被忽略的点：文件句柄泄漏。即使 pipeline 结束，fs.createReadStream 的底层 fd 不一定立即释放，尤其在出错时。稳妥做法是显式监听 finish 和 error，并在回调里手动 readStream.destroy()（尽管 pipeline 通常已做，但多一层保险无害）。
复杂点在于：当 pipeline 中嵌套了异步操作（如数据库查表补字段），这些操作无法被 pipeline 自动取消。必须用 AbortController 手动中止，并确保每个异步步骤都响应 signal.aborted ——这点几乎总是被跳过。
