Node.js 流(Stream)

流是 Node 最强大也最被低估的抽象。它解决一个核心问题:如何处理"大到放不下内存"的数据。一个 5GB 的日志文件、一段正在下载的视频、一个实时音频流——这些都不可能一次性塞进内存。流让你"分块读取、边读边处理",内存占用始终恒定。Node 的 fs、http、网络、压缩、加密模块,底层全是流。

1. 为什么需要流?

假设你要读一个 1GB 的文件。用 readFile 会把整个文件加载到内存,直接内存爆掉:

// readFile 会把整个文件加载进内存
// 文件 1GB → 内存吃 1GB,大文件直接 OOM(内存溢出)
const fs = require("fs");
fs.readFile("huge.mp4", (err, data) => {
  // data 是个 1GB 的 Buffer,危险!
});

// 流(Stream)是"分块处理"的抽象
// 一边读一边处理,内存只占一小块
const readStream = fs.createReadStream("huge.mp4");
// 默认每块 64KB(highWaterMark),处理完再读下一块
// 1GB 的文件,内存峰值可能就几 MB

流的本质是"分块处理"。它把数据切成一小块一小块(默认 64KB),读一块、处理一块、丢掉,再读下一块。1GB 文件用流处理,内存峰值可能就几 MB——这是 Node 能扛高并发、处理大文件的关键。

2. 四种流类型

Node 把流分成四类,记住它们就能理解所有流式 API:

// Node 有四种流

// 1. Readable(可读流):只能读
//    如 fs.createReadStream、HTTP 请求体 req
const { Readable } = require("stream");
const r = new Readable({
  read() {
    this.push("数据块");
    this.push(null);   // null 表示读完
  }
});

// 2. Writable(可写流):只能写
//    如 fs.createWriteStream、HTTP 响应体 res
const { Writable } = require("stream");
const w = new Writable({
  write(chunk, encoding, callback) {
    console.log("写入:", chunk.toString());
    callback();
  }
});

// 3. Duplex(双工流):可读又可写(读写独立,像电话)
//    如 TCP socket

// 4. Transform(转换流):读进来变换后写出去
//    如 zlib(压缩)、crypto(加密)

真实世界的例子:fs.createReadStream 是 Readable,fs.createWriteStream 是 Writable,TCP socket 是 Duplex,压缩/加密是 Transform。HTTP 的 req 是 Readable,res 是 Writable。

3. 可读流的两种模式

可读流有两种工作模式,这是新手最容易困惑的点:

默认是暂停模式,监听 data 会自动切到流动模式:

const fs = require("fs");
const stream = fs.createReadStream("note.txt", "utf8");

// data:每读到一块就触发(自动进入流动模式)
stream.on("data", (chunk) => {
  console.log("读到:", chunk.length, "字符");
});

// end:所有数据读完后触发
stream.on("end", () => {
  console.log("读完了");
});

// error:出错时触发
stream.on("error", (err) => {
  console.error("读流出错:", err);
});

// 也可以手动控制读取(暂停/恢复)
// stream.pause();
// stream.resume();

常用事件:data(读到一块)、end(读完)、error(出错)、close(流关闭)。可读流还能 pause() / resume() 控制读取节奏。

4. pipe:管道的魔法

pipe 把可读流"接"到可写流,数据自动从前者流向后者。这是流最常用的 API:

const fs = require("fs");

// pipe:把可读流接到可写流,自动处理数据传递
const readStream = fs.createReadStream("source.txt");
const writeStream = fs.createWriteStream("copy.txt");

readStream.pipe(writeStream);

writeStream.on("finish", () => {
  console.log("复制完成");
});

// 经典场景:HTTP 响应直接返回文件
// const http = require("http");
// http.createServer((req, res) => {
//   fs.createReadStream("big.pdf").pipe(res);   // 边读边发
// }).listen(3000);
// 内存占用极低,适合大文件下载

pipe 的威力在于可以链式拼接——读文件 → 解压 → 解密 → 写文件,一行代码搞定。而且 pipe 自动处理背压(下面讲),这是它比手动监听 data 更安全的原因。

5. Transform:转换流

Transform 流既可读又可写,但它的作用是"读进来 → 变换 → 写出去"。压缩、加密、格式转换都是这个套路:

const { Transform } = require("stream");
const fs = require("fs");

// 自定义转换流:把数据变大写后传出
const upper = new Transform({
  transform(chunk, encoding, callback) {
    // chunk 是 Buffer,转成大写后 push 出去
    const result = chunk.toString().toUpperCase();
    this.push(result);
    callback();
  }
});

// 串起来:读文件 → 转大写 → 写文件
fs.createReadStream("in.txt")
  .pipe(upper)
  .pipe(fs.createWriteStream("out.txt"));

// 实战:边读边压缩边写
// const zlib = require("zlib");
// fs.createReadStream("log.txt")
//   .pipe(zlib.createGzip())              // gzip 压缩(转换流)
//   .pipe(fs.createWriteStream("log.txt.gz"));

Node 内置的 zlib(压缩)、crypto(加密)都提供 Transform 流。配合 pipe,可以做到"边读边压缩边写",全程几乎不占内存——这是处理大文件的金科玉律。

6. 背压(backpressure):核心难点

这是流最关键也最容易被忽视的概念。生产者和消费者速度不匹配时怎么办?比如读文件很快(磁盘读取),但写文件很慢(目标磁盘慢或网络上传)。如果一味往下游塞数据,下游缓冲区会爆,内存飙升——这就是"背压"问题。

// 背压(backpressure):生产快、消费慢的问题
const fs = require("fs");
const src = fs.createReadStream("fast.txt");
const dst = fs.createWriteStream("slow.txt");

// ❌ 危险写法:手动处理 data,生产比消费快会撑爆内存
src.on("data", (chunk) => {
  const ok = dst.write(chunk);
  if (!ok) {
    // 写不下了,要暂停读取
    src.pause();
    dst.once("drain", () => src.resume());
  }
});

// ✅ 推荐写法:用 pipe,自动处理背压
src.pipe(dst);
// pipe 内部就是上面的逻辑,写不下了会自动暂停读取

// 这就是为什么永远推荐用 pipe,而不是手动监听 data

正确处理背压的逻辑是:当下游 write 返回 false,说明它暂时写不动了,要暂停上游读取;等下游触发 drain 事件(消化完了),再恢复读取。手写这套逻辑很繁琐——所以永远优先用 pipe 或 pipeline,它们内部自动处理背压。

7. 现代 API:异步迭代与 pipeline

Node 10+ 支持用 for await...of 异步迭代消费流,代码极其优雅。同时 stream/promises 的 pipeline 比传统 pipe 更安全(自动清理、错误传播):

// 现代 Node:用异步迭代消费流(最优雅)
const fs = require("fs");

async function readFileByChunks(path) {
  const stream = fs.createReadStream(path, "utf8");
  let total = 0;
  for await (const chunk of stream) {
    total += chunk.length;
  }
  console.log("总共:", total, "字符");
}

readFileByChunks("big.txt");

// pipeline:替代 pipe,更好地处理错误和清理
const { pipeline } = require("stream/promises");
const zlib = require("zlib");

async function compress() {
  await pipeline(
    fs.createReadStream("log.txt"),
    zlib.createGzip(),
    fs.createWriteStream("log.txt.gz")
  );
  console.log("压缩完成");
}
compress();

为什么推荐 pipeline 而不是 pipe:pipe 在源流或目标流出错时,不会自动销毁另一个,容易导致内存泄漏;pipeline 会正确清理所有流,并把错误通过 Promise reject 出来。生产代码请用 pipeline。

8. 实战:大文件压缩上传

const fs = require("fs");
const zlib = require("zlib");
const { pipeline } = require("stream/promises");
const http = require("http");

async function compressAndUpload() {
  // 边读、边压缩、边上传到服务器
  await pipeline(
    fs.createReadStream("big.log"),       // 读
    zlib.createGzip(),                    // 压缩
    // 上传到目标 HTTP 服务器(请求体是可写流)
    http.request({
      host: "backup.example.com",
      path: "/upload",
      method: "POST",
      headers: { "Content-Encoding": "gzip" }
    })
  );
  console.log("压缩并上传完成");
}

compressAndUpload().catch(console.error);

这段代码处理一个 10GB 的日志文件,内存占用可能就几 MB——这就是流的威力。如果用 readFile,机器早就 OOM 了。

小结

流是 Node 的精华。能流式处理的就别一次性读全——内存稳定、性能更好。记住四点:四种流类型(读写双转)、pipe 自动处理背压、pipeline 比pipe 更安全、for await...of 是现代消费姿势。下一篇我们深入 Node 的异步编程——事件循环的细节。

← 上一篇 Node.js 文件系统

下一篇 Node.js 异步编程 →

✈️💬