7.4 管道（pipe）机制与手动流控实现

在 7.3 节我们已经看到，当可读流产生数据速度持续快于可写流消费速度时，内部缓冲区会不断膨胀，最终导致内存压力甚至进程崩溃。Node.js 通过 管道（pipe）机制 优雅地解决了这个问题，同时也提供了一套手动流控的方法。本节我们详细探讨 pipe() 的工作原理、使用方式，以及如何在不依赖 pipe() 的情况下手动实现带背压控制的流对接。

7.4.1 pipe() 的基本用法与原理

pipe() 是可读流暴露的核心方法，用于将可读流的输出自动导向可写流。它的语法极其简洁：

readable.pipe(writable);

内部，pipe() 自动完成了以下工作：

监听可读流的 data 事件，每收到一个数据块就调用可写流的 write() 写入。
根据 write() 的返回值进行背压控制：如果 write() 返回 false（表示写入缓冲区已满或达到 highWaterMark），则立即暂停可读流（调用 readable.pause()），等待可写流排空。
监听可写流的 drain 事件，当缓冲区排空后恢复可读流（调用 readable.resume()），从而继续读取数据。
传播关闭/结束事件：当可读流结束时，调用可写流的 end() 方法；当任一流发生错误或过早关闭时，执行必要的清理（在较早版本中 pipe() 不会自动处理错误销毁另一个流，需通过 stream.pipeline 保证）。

这些细节都被封装在 Node.js 的源码中，开发者无需关心背压调控，只需一行 pipe() 即可安全地在流与流之间传输任意大小的数据。

7.4.2 pipe() 的实际应用：大文件复制

想象一下，我们需要复制一个 2GB 的文件。如果使用 fs.readFile 将整个文件读入内存再写入，内存会瞬间暴涨，甚至导致进程 OOM。而利用 pipe()，我们可以用极低的内存占用完成同样的任务：

const fs = require('fs');

const readStream = fs.createReadStream('source.dat');
const writeStream = fs.createWriteStream('target.dat');

readStream.pipe(writeStream);

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

运行这段代码时，数据会分块从源文件流向目标文件，每一时刻内存中只有少量数据块，背压完全由 pipe() 自动调节：当目标磁盘写入慢时，读流自动暂停；写入完成后自动恢复读取。

7.4.3 链式调用与错误处理

pipe() 返回的是目标流（Writable），因此可以链式调用，将多个流串联起来：

fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('output.txt.gz'));

常见需求还包括对流的转换，如上述的压缩。但 pipe() 的错误处理存在一个陷阱：它不会自动销毁管道中的其他流，也不会将错误从一个流传播到另一个流。例如，如果上面的压缩过程中出现错误，读流和写流可能并未被关闭，导致文件句柄泄漏或不完整输出。

早期社区使用 pump 或 pipeline 来解决此问题。Node.js 从 10.0 版本起提供了 stream.pipeline() 函数，它会在任一流传出错误时自动销毁管道中的所有流，并调用最终回调。 推荐在任何新代码中使用 stream.pipeline：

const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');

pipeline(
  fs.createReadStream('input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('output.txt.gz'),
  (err) => {
    if (err) {
      console.error('管道失败', err);
    } else {
      console.log('管道成功完成');
    }
  }
);

stream.pipeline 还支持 Promise 封装（Node.js 15+），使得 async/await 风格写法成为可能：

const { pipeline } = require('stream/promises');

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

7.4.4 手动流控：当你不使用 pipe()

虽然 pipe() 和 pipeline() 已经足够大部分场景，但有些情况下我们需要更精细的控制，例如：

需要在数据流通过程中做额外处理（修改数据、记录日志、自定义背压策略）
需要将数据分发到多个目标流
在流对接过程中需要动态插入或移除流的环节

这时就需要实现手动流控，自己处理 data、drain、end 等事件，正确执行背压。

基本步骤如下：

监听可读流的 data 事件。
在 data 回调中，调用可写流的 write()。
检查 write() 返回值：若为 false，则暂停可读流。
监听可写流的 drain 事件，一旦触发则恢复可读流。
监听可读流的 end 事件，调用可写流的 end()。
注意错误处理：监听 error 事件，必要时关闭两个流。

下面是一个手动流控的例子，将一个文件数据写入另一个文件，并在每次写入时打印进度：

const fs = require('fs');

const reader = fs.createReadStream('bigfile.mp4');
const writer = fs.createWriteStream('copy.mp4');
let totalBytes = 0;

reader.on('data', (chunk) => {
  const canContinue = writer.write(chunk);
  totalBytes += chunk.length;
  console.log(`已写入 ${(totalBytes / 1024 / 1024).toFixed(2)} MB`);

  // 如果写入缓冲已满，暂停读取
  if (!canContinue) {
    reader.pause();
    console.log('背压触发，读流暂停');
  }
});

// 可写流排空后恢复读取
writer.on('drain', () => {
  console.log('缓冲区排空，读流恢复');
  reader.resume();
});

reader.on('end', () => {
  writer.end();
  console.log('读取完毕，写流结束');
});

// 错误处理
reader.on('error', handleError);
writer.on('error', handleError);

function handleError(err) {
  reader.destroy();
  writer.destroy();
  console.error('流发生错误', err);
}

这个示例清楚地展现了 pipe() 内部的背压逻辑：通过 pause() / resume() 与 write() 返回值配合，实现数据流的自然调速。手动实现虽然增加了代码量，但给予了开发者完全控制每个字节流动的能力。

7.4.5 手动流控中的常见陷阱

忘记暂停：如果 write() 返回 false 却没有暂停可读流，可读流继续发射 data 事件，造成缓冲区无限增长，最终可能导致内存溢出。这正是 pipe() 自动化避免的情况。
没有监听 drain：暂停后如果不恢复，流会永远卡住。
没有处理 end：可读流结束后，如果不调用可写流的 end()，目标文件可能会缺少尾部数据或一直处于打开状态。
错误传播不彻底：一个流出错后没有销毁另一个流，导致句柄泄漏。建议使用辅助函数或封装 pipeline。

7.4.6 管道机制的内部实现简析

理解源码有助于在复杂场景下调试。Node.js 的实现（简化版）大致如下：

Readable.prototype.pipe = function(dest, options) {
  const src = this;

  // 监听数据
  src.on('data', function(chunk) {
    const ret = dest.write(chunk);
    if (!ret) {
      src.pause();
    }
  });

  // 监听可写流drain，恢复可读流
  dest.on('drain', function() {
    src.resume();
  });

  // 结束时结束写流
  src.on('end', function() {
    dest.end();
  });

  // 错误处理（简化）
  src.on('error', err => {
    dest.destroy(err);
  });
  dest.on('error', err => {
    src.destroy(err);
  });

  // ... 更多事件处理
  return dest;
};

真实源码还会处理流的关闭时机、选项传递、备份事件等，但核心逻辑正如上述过程。

7.4.7 管道机制与背压控制的实践建议

优先使用 stream.pipeline 或 pipe()，它们在 99% 的场景下都能正确工作，代码简洁，且背压控制经过充分验证。
仅在需要特殊逻辑时手动流控，并牢记背压三要素：write() 返回值、pause()、drain 事件恢复。
始终处理 error 事件，避免未捕获异常导致进程崩溃。
考虑使用 stream.finished() 做结束监控，它返回 Promise，可以可靠地等待流完成或错误发生。

通过掌握管道机制的原理，以及手动控制流的能力，开发者可以利用 Node.js 流模型构建高效、内存友好的数据处理通道，无论是处理大文件、网络代理、还是实时日志传输，都能做到游刃有余。