一、为什么流是Node.js处理大数据的正确姿势

很多初学者处理文件时习惯用fs.readFile一次性把内容读进内存，几十KB的配置文件的确认方便。但换成2GB的日志文件、持续上传的视频流、或者每秒数千条的实时数据，这种方式会直接把内存打爆。

Node.js的Stream（流）解决的正是这个问题：它不把数据当作一整块，而是切成一个个chunk（数据块）逐步消费。数据像水流一样从源头流向目的地，内存中始终只保留当前正在处理的那一小段。这就是流式处理的核心价值——用时间换空间，把峰值内存从"整个数据集"降到"单个chunk"。

这里有个容易被忽略的点：流并不是Node.js独有的概念，Unix管道|就是最古老的流式思想。Node.js把它抽象成了EventEmitter的实例，让数据流动过程可以被监听、暂停、恢复和转发。理解这一点，后面的pipe和背压就顺理成章了。

经验提示：判断该不该用流，看一个指标——数据规模是否可能超过可用内存，或者数据是否是逐步到达的。只要满足其一，就应该考虑流式方案。

二、四种流类型，各管一段职责

Node.js中的流分为四类，它们不是并列关系，而是层层递进的能力叠加。

Readable（可读流）是数据的生产者。它从文件、网络socket、标准输入等源头持续产出数据。消费方式是监听data事件或调用read()，也可以直接pipe给下游。

Writable（可写流）是数据的消费者。它接收chunk并写入目标，比如文件、HTTP响应、标准输出。关键方法是write()和end()。

Duplex（双工流）同时具备读写能力，但读写两条通道是独立的，互不干扰。TCPsocket就是典型例子——你可以同时往连接里写请求和读响应。

Transform（转换流）是Duplex的特例，它的特殊之处在于：写进去的数据经过处理后，会变成可读侧的输出。输入和输出存在转换关系，压缩、加密、大小写转换都属于这一类。

类型	方向	典型场景	关键方法
Readable	只读	读文件、接收请求体	read()、push()
Writable	只写	写文件、发送响应	write()、end()
Duplex	读写独立	TCPsocket	同时具备两者
Transform	读写联动	gzip、加密、转码	transform()

三、流事件：数据流动的观测点

流继承自EventEmitter，整个生命周期通过事件暴来。掌握这些事件，才能做出正确的错误处理和流程控制。

data在可读流有数据可读时触发，是最常用的消费入口。end表示可读流数据已经读完，不会再有新数据。error在任意阶段出错时触发，这是必须处理的，否则进程可能崩溃。finish在可写流的end()调用后、所有数据刷盘完成时触发。close表示底层资源已经关闭，它和finish不是一回事，finish关注数据写完，close关注句柄释放。

反思：很多人只监听finish就宣布任务完成，但文件句柄可能还没释放。在需要立即重命名或删除文件的场景，应该等close事件，否则在部分平台上会遇到EBUSY或权限错误。

四、可读流：从手动push到pipe输出

下面用代码号学习编程的方式创建一个自定义可读流，手动往里面推数据，再通过管道输出到控制台。

const { Readable } = require('stream');

// 代码号学习编程：创建自定义可读流
const inStream = new Readable({
  read() {}
});

inStream.push('代码号学习编程: ');
inStream.push('专注编程技术教程');
// push(null) 表示数据源已枯竭
inStream.push(null);

// 将流内容转发到标准输出
inStream.pipe(process.stdout);

输出：

代码号学习编程: 专注编程技术教程

这里的read(){}是空实现，因为我们采用手动push模式。push(null)是流的终止信号，缺少它，下游会一直等待，程序无常结束。pipe(process.stdout)把可读流接到标准输出这个可写流上，数据自动流动。

如果换成真实文件，写更简洁：

const fs = require('fs');
fs.createReadStream('./access.log').pipe(process.stdout);

五、可写流：接管标准输入

可写流的自定义重点在write(chunk,encoding,callback)方法。处理完一个chunk后必须调用callback()，否则流会认为写入未完成，进而阻塞后续数据。

const { Writable } = require('stream');

const outStream = new Writable({
  write(chunk, encoding, callback) {
    console.log('收到: ' + chunk.toString().trim());
    callback(); // 通知流：这块处理完了
  }
});

process.stdin.pipe(outStream);

运行后输入任意内容并回车，控制台会回显。callback的参数可以传错误对象，用于向上游传递写入失败信号，这一点在对接数据库或远程存储时很关键。

六、双工流与转换流：读写如何协作

双工流的读写互不干扰，下面的例子中，写入的内容走write通道，读取的内容走read通道。

const { Duplex } = require('stream');

const duplexStream = new Duplex({
  read() {
    this.push('读取通道：数据准备就绪\n');
    this.push(null);
  },
  write(chunk, encoding, callback) {
    console.log(`写入通道：${chunk.toString()}`);
    callback();
  }
});

duplexStream.write('代码号学习编程');
duplexStream.on('data', (chunk) => {
  console.log(chunk.toString());
});

输出：

写入通道：代码号学习编程
读取通道：数据准备就绪

转换流则把输入和输出绑定起来，下面的例子实现大写转换：

const { Transform } = require('stream');

const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    // 第一个参数传 null 表示无错误
    callback(null, chunk.toString().toUpperCase());
  }
});

upperCaseTransform.on('data', (chunk) => {
  console.log(chunk.toString());
});

upperCaseTransform.write('hello ');
upperCaseTransform.write('code number');
upperCaseTransform.end();

输出：

HELLO CODE NUMBER

transform里的callback(null,result)是固定套路：第一个参数留给错误，第二个参数是要推给可读侧的数据。忘记传第二个参数，下游就收不到任何东西，这是新手常踩的坑。

七、pipe管道与链式处理

pipe()把可读流和可写流连接起来，同时自动处理背压——当下游写入慢时，它会暂停上游的读取。这是它比自己监听data手动write更值得用的原因。

文件复制是最基础的管道应用：

const fs = require('fs');

const readableStr = fs.createReadStream('./input.txt');
const writableStr = fs.createWriteStream('./output.txt');

readableStr.on('error', (err) => console.error('读取错误:', err.message));
writableStr.on('error', (err) => console.error('写入错误:', err.message));
writableStr.on('finish', () => console.log('文件复制完成'));

readableStr.pipe(writableStr);

当input.txt不存在时，会看到读取错误:ENOENT:nosuchfileordirectory；存在时则输出文件复制完成。

链式处理把多个转换流串起来，数据经过每一级加工后继续往。下面实现读取→gzip压缩→写出的完整链路：

const fs = require('fs');
const zlib = require('zlib');

fs.writeFileSync('source.txt', '代码号学习编程：流式压缩示例');

fs.createReadStream('source.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('source.txt.gz'))
  .on('finish', () => console.log('压缩完成'));

这里.pipe()返回的仍是流，所以可以继续链式调用。中间环节必须是Transform或Duplex，因为它既要接收上游数据，又要向下游输出。

八、背压：流式系统的自我调节

背压（Backpressure）是流机制里最容易被忽视、却最能体现设计功力的部分。它的含义是：当消费者处理速度跟不上生产者时，系统通过反向信号让生产者减速，避免数据在内存中堆积。

下面的例子中，可写流每个chunk延迟1秒处理，而可读流瞬间就能产出全部数据。pipe()会自动感知这个速度差并施加背压。

const { Readable, Writable } = require('stream');

const readable = Readable.from(['数据 1', '数据 2', '数据 3']);

const writable = new Writable({
  write(chunk, encoding, callback) {
    setTimeout(() => {
      console.log(`已处理: ${chunk.toString()}`);
      callback();
    }, 1000);
  }
});

readable.pipe(writable);

输出会以每秒一条的节奏出现：

已处理: 数据 1
已处理: 数据 2
已处理: 数据 3

callback()在这里扮演了关键角色：它没被调用之前，可写流不会请求下一个chunk，上游自然就停住了。如果绕过pipe手动监听data并write，就必须自己根据write()的返回值判断是否需要pause()和resume()，复杂度高很多。

个人建议：能用pipe()就不要手写数据转发。只有在需要精细控制流速、或者做多路分流时，才考虑手动管理背压。现在Node.js还可以用stream.pipeline()替代裸pipe，它能在任意环节出错时统一销毁所有流，避免资源泄漏。

九、流式方案带来的实际收益

把流用到项目里，最直接的感受是内存曲线变平了。处理1GB文件时，一次性读取会让RSS冲到1GB以上，而流式处理通常稳定在几十MB。是响应变快：数据边到达边处理，首字节时间明显缩短，这对HTTP代理、日志采集、视频转码这类场景尤其重要。

另一层价值是可组合性。source.pipe(transform1).pipe(transform2).pipe(destination)这种写法本身就是对处理流程的声明式描述，阅读代码时数据流向一目了然。压缩、加密、校验、转码这些操作可以像积木一样拼接，而不必写成一坨回调嵌套。

十、本节课程知识要点

• 流的核心是分块处理，用时间维度换内存空间，适合大数据集和逐步到达的数据。

• 四种类型各有分工：Readable产数据，Writable收数据，Duplex双向独立，Transform读写联动。

• push(null)结束可读流，callback()结束一次写入，callback(null,data)推送转换结果，这三个是自定义流的关键动作。

• pipe()不只是连接，它内置了背压管理，比手动转发更可靠。

• 链式处理要求中间流是Transform或Duplex，因为只有它们能同时充当上下游。

• 错误处理不能只依赖finish，在需要释放资源的场景要关注close；生产环境推荐用stream.pipeline()统一管理生命周期。

十一、练习与思考

1. 用Transform实现一个流，把输入中的敏感手机号中间四位替换为****，再pipe到文件。

2. 改造背压示例，不用pipe()，改用data事件加write()返回值手动实现暂停与恢复，体会两者的差异。

3. 思考：为什么Duplex的读写通道要独立设计，而不是像Transform那样联动？结合TCP半关闭状态说明。

4. 尝试用stream.pipeline()重写gzip压缩示例，并故意在中途传入一个不存在的文件，观察错误是如何传播到回调的。

十二、常见问题排查

遇到writeafterend错误，通常是对已经调用end()的流继续write()，检查数据源是否重复触发。遇到ERR_STREAM_PREMATURE_CLOSE，多半是下游提前关闭而上游还在写，需要在管道上补充错误监听。至于进程不退出，常见原因是可读流没有push(null)，或者忘记消费可写流的finish事件。这些问题在开发阶段用stream.finished()辅助监听状态，能省下不少排查时间。