Node.js Stream

Node.js Stream 的四种类型、pipe 机制、背压与 highWaterMark 的工作原理。

#concept #growing #backend #nodejs #javascript

[!info] related notes

所属 MOC: Node.js MOC

上位概念: Stream

相关概念: Node.js Buffer, Node.js 事件循环阶段

Node.js Stream

一句话定义

Stream 是 Node.js 处理流式数据的核心抽象，让你不用一次性把全部数据加载到内存，而是分块读写。

四种流类型

类型	说明	典型场景
Readable	可读流，数据源	`fs.createReadStream`, `http.IncomingMessage`
Writable	可写流，数据目的地	`fs.createWriteStream`, `http.ServerResponse`
Duplex	可读可写，双工	`net.Socket`, TCP 连接
Transform	读写+转换	`zlib.createGzip`, `crypto.createCipheriv`

本页关注 Node.js 具体 API；流式处理的一般动机见 stream。

基本用法

以下示例使用 ESM，保存为 .mjs；input.txt 必须存在，输出文件会被创建或覆盖。

import fs from 'node:fs';
import { pipeline } from 'node:stream/promises';
import zlib from 'node:zlib';

await pipeline(
  fs.createReadStream('input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('input.txt.gz')
);

pipeline 完成表示本次管道处理完成；压缩正确性可通过解压后与输入比较验证。它会协调错误与销毁流，但不会自动回滚已经写入的文件或外部副作用。

pipe 与背压

pipe() 的核心价值是自动管理背压（backpressure）。

背压发生场景：可读流产生数据的速度 > 可写流消费数据的速度。

当 writable.write(chunk) 返回 false 时，该 chunk 已被接收；调用方应暂停后续写入，等待 drain 后继续。继续写入会让队列增长。readable.pipe(writable) 会协调暂停与恢复，但不会替你完整处理所有流的错误和关闭。

highWaterMark

highWaterMark 是缓冲阈值，不是硬内存上限。字节模式通常按字节计，object mode 按对象数计；经 setEncoding() 解码的可读流还有字符计数差异。默认值与流类型、Node 版本有关，不应统一记为 64 KB。

import fs from 'node:fs';

const readable = fs.createReadStream('file.txt', {
  highWaterMark: 16 * 1024 // 本例显式设置 16 KiB
});

可写端的队列达到阈值时，write() 可返回 false。单个大 chunk、对象本身大小、多个流缓冲和应用层积压都会让实际内存高于这个阈值。

自定义 Transform 流

下面只处理 ASCII 字节。任意 UTF-8 文本可能跨 chunk 拆开多字节字符，不能逐块 toString() 后假定得到完整字符；需要文本转换时可用 StringDecoder 或先正确设置解码层。

import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

const upperAscii = new Transform({
  transform(chunk, encoding, callback) {
    const out = Buffer.from(chunk);
    for (let i = 0; i < out.length; i++) {
      if (out[i] >= 0x61 && out[i] <= 0x7a) out[i] -= 32;
    }
    callback(null, out);
  }
});

await pipeline(process.stdin, upperAscii, process.stdout);

输入 abc 输出 ABC；非 ASCII 字节保持原样，不提供 Unicode 大小写转换。

流的错误处理

Promise 版 pipeline 出错会 reject；调用方必须处理失败，CLI 还应设置非零退出码。不能只打印错误后以成功状态结束。

pipeline 通常会销毁失败链条中的流。用于 HTTP 等场景时，这可能在发送自定义错误响应前关闭连接；复用流或需要特定恢复策略时应检查 API 边界。

何时使用 Stream

大文件处理：读写大文件，避免一次性加载；保持背压且各阶段不累积全部数据时可限制内存增长
网络传输：HTTP 请求/响应天然是流
数据转换链：gzip、加密、CSV 解析等管道式处理
实时数据：日志、传感器数据等持续到达的数据源

常见误区

不要忽略背压：不处理背压会导致内存暴涨
pipe 不传播错误：用 pipeline 代替
stream.destroy() 释放资源：异常退出时记得调用
可以在对象模式和字节模式之间转换，但必须有显式适配（如对象序列化）；不能直接向字节 writable 写任意对象

学习检查

write() 返回 false 时，刚传入的 chunk 是否需要重发？
为什么设置 highWaterMark 并不能保证进程内存不超限？
为什么逐 chunk 解码 UTF-8 可能损坏字符？

参考

Node.js Stream API