Node.js Streams 实战：背压、pipeline 与大文件处理

📅 2026-10-19 📂 后端 🏷️ Node.js, Streams, 背压, 性能
7 分钟阅读 · 1,858 字

处理大文件、HTTP 上传、压缩和导出时，最容易踩的坑不是 API 太复杂，而是把“流式”误当成“自动省内存”。如果生产者持续写入而消费者处理较慢，队列仍会增长；如果代码把所有块拼成数组，流接口也救不了内存。本篇用 Node.js 内置 Streams API 把数据流、背压、错误、取消和可验证性串起来。

文中示例使用 Node.js 内置模块，不依赖第三方包。上线前要按项目支持的 Node 主版本运行测试；本文不假定不同版本的实验性 API 可互换。

1. 四种流与队列

  • Readable 提供数据，例如 fs.createReadStream()、HTTP IncomingMessage。
  • Writable 消费数据，例如 fs.createWriteStream()、HTTP ServerResponse。
  • Duplex 同时可读写，例如 TCP socket。
  • Transform 是带转换逻辑的 Duplex，例如 gzip、压缩或序列化。

每一段流都有自己的缓冲。highWaterMark 是流开始施加压力的阈值，不是硬性内存上限：一个大 chunk 可以让实际队列超过阈值；对象模式按对象数量计数，却不考虑每个对象的真实字节大小。一个 Transform 同时有可写和可读队列，因此整个管线的内存还要加上各段缓存、当前处理中数据、应用层数组、数据库游标预取和压缩器内部状态。

对于 Writable，write(chunk) 返回 false 表示当前应该停止继续写，等待 'drain' 后再写；这不代表当前 chunk 被丢弃。对 Readable 的手动 push()，当它返回 false 时，生产者也应暂停，等消费者继续读取。多数业务代码应优先用 pipeline()，让管线在各段之间传递背压并集中处理失败和销毁。

2. 用 pipeline 把慢消费者纳入节奏

下面的例子逐条生成记录，Transform 将对象变成 NDJSON，慢速 Writable 只统计输出。数据不会先被收集成一个大数组。把 Writable 的 highWaterMark 设得很小，是为了让测试容易触发背压，不是推荐所有生产配置照抄这个数值。

const { Transform, Writable } = require('node:stream');
const { pipeline } = require('node:stream/promises');

async function* createRows({ signal }) {
  for (let id = 1; id <= 1000; id += 1) {
    if (signal.aborted) throw signal.reason ?? new Error('aborted');
    yield { id, name: `user-${id}` };
  }
}

async function main() {
  let rows = 0;
  let bytes = 0;
  let maxBufferedBytes = 0;

  const toNdjson = new Transform({
    writableObjectMode: true,
    readableHighWaterMark: 64,
    transform(row, _encoding, callback) {
      rows += 1;
      callback(null, `${JSON.stringify(row)}\n`);
    },
  });

  const slowSink = new Writable({
    highWaterMark: 1,
    write(chunk, _encoding, callback) {
      bytes += chunk.length;
      maxBufferedBytes = Math.max(maxBufferedBytes, this.writableLength);
      setTimeout(callback, 0);
    },
  });

  const controller = new AbortController();
  await pipeline(createRows, toNdjson, slowSink, { signal: controller.signal });
  console.log(`rows=${rows} bytes=${bytes} maxBufferedBytes=${maxBufferedBytes}`);
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});

pipeline() 在任一阶段报错时会拒绝 Promise 并清理整条管线。把请求取消、任务超时或客户端断开映射到 AbortController，再将 signal 传入 pipeline；异步生成器也必须响应它，不能在取消后继续查询或产出记录。示例中的检查保证下次迭代时停止；如果生成器正在等待数据库、网络或计时器，还要把 signal 传进该操作并确认底层 API 会中止。

3. 什么时候要自己处理 drain

只有在确实需要逐块写入自定义 Writable 时才手动管理背压：write() 返回 false 后暂停生产，在 'drain' 到达后恢复，并同时处理 'error'、提前 'close' 与取消。仅仅 await once(writable, 'drain') 可能在流已关闭且永远不会再触发 drain 时悬挂；需要把 drain 等待和 stream 完成/失败状态一起管理。

最常见的反例是：

async function badCopy(source, destination) {
  for await (const chunk of source) {
    destination.write(chunk);
  }
}

这段代码忽略 write() 的返回值，生产速度一旦超过目的端，Writable 缓冲会不断扩大。若每个 chunk 都启动一个未等待的异步任务，积压会从流缓冲扩散到 Promise、回调和闭包里。增加 highWaterMark 只能推迟压力出现，不能解决无限制生产。

4. 字节边界不是消息边界

Readable 发出的 chunk 由运行时和底层 I/O 决定，不承诺刚好对应一行 JSON、一个 HTTP 事件或一个 UTF-8 字符。解析文本时使用 setEncoding('utf8') 或 StringDecoder 处理多字节字符跨 chunk 的情况；解析 NDJSON、SSE 等协议时，还要保留未完成片段，按协议分隔符切出完整消息，并对单条消息长度设置上限。

不要把任意 chunk 直接 JSON.parse()，也不要默认一次 data 事件就是一次业务消息。反过来，使用 stream.toArray()、Buffer.concat() 或 chunks.join('') 会把数据整体保留在内存中，除非输入大小有明确上限且已做容量预算。

5. HTTP 上传、下载与文件导出

Node HTTP 的请求体本身是 Readable，响应是 Writable。理论上可以把请求流接到校验、限长、哈希、临时文件或对象存储管线，但真实框架的中间件可能已先缓存或解析整个请求体；先检查 body parser、multipart 解析器与反向代理的行为，不能仅凭 Controller 使用 pipeline() 就宣称端到端流式。

  • 上传应同时设置代理、HTTP 解析器、应用层与存储层的请求大小限制；验证身份、授权、声明 MIME 与实际文件内容，使用随机存储名和隔离暂存目录。
  • 读取完毕并校验成功后再提升为正式文件；失败、取消和超时都要删除临时对象，避免孤儿文件。
  • 下载或导出要在开始发送响应前完成授权和必要的元数据校验。响应头一旦发出，流中途失败时通常不能再改成 JSON 错误码，应记录脱敏错误并结束连接。
  • 数据库导出优先使用游标或分页生成器；同时限制并发任务、单任务时长和每个租户配额，避免多个“内存安全”的流任务相乘后拖垮进程。

流式传输不等于安全上传，也不等于零拷贝。文件内容检查、压缩炸弹防护、磁盘配额、权限、备份和恢复仍是独立控制面。

6. 观测与测试清单

定位积压时观察 readableLength、writableLength、readableHighWaterMark、writableHighWaterMark、处理吞吐、事件循环延迟和 RSS。不要把单次内存快照当作泄漏结论；要在稳定输入速率、慢消费者、错误、取消和并发压力下观察趋势。

至少验证以下场景：

  1. 慢速 Writable 下，输出内容正确且队列没有随输入总量线性增长。
  2. 任一 Transform 抛错后，来源和目的端都结束，任务拒绝且临时文件清理。
  3. 客户端断开或 AbortSignal 触发后，底层读取、数据库游标和写入任务都停止。
  4. chunk 在 UTF-8 多字节字符、换行和协议分隔符中间断开时，解析结果仍正确。
  5. 超大单条记录、恶意长度声明、并发上传和磁盘写满会被限制并给出可诊断结果。

官方资料

先用内置 pipeline() 建立正确的压力、错误与取消传播，再通过有代表性的压测调整缓冲阈值。任何测试结论都要注明 Node 版本、数据规模、目的端速度和并发数；一次本地演示不能证明生产峰值安全。

Live Notes

读者弹幕

留言提交后会先进入审核，审核通过后对其他读者公开。