什么是流（streams）中的背压（backpressure），如何处理它？

Question

Accepted Answer

**背压**是指数据被**生产的速度比消费速度更快**时发生的情况。在流中，如果一个快速的可读源向一个慢速的可写目标推送数据，未消费的数据会在内存中缓存——如果不进行处理，缓冲区会不断增长，直到内存耗尽。背压是保持生产者和消费者平衡的机制（和原则）。

## 问题

```text
Fast source ──(1 GB/s)──▶ Slow destination (10 MB/s write speed)
                          ▲
             The 990 MB/s difference piles up in a memory buffer → OOM crash
```

示例：快速读取一个大文件并将其写入一个慢速的网络套接字或磁盘——如果你不尊重目标的速度，内存会急剧增长。

## pipe 自动处理背压

```js
import { createReadStream, createWriteStream } from "fs";

// ✅ pipe automatically pauses the source when the destination is full,
//    and resumes it when the destination drains — backpressure handled for you
createReadStream("huge.file").pipe(createWriteStream("copy.file"));
```

这是使用 `pipe`/`pipeline` 的主要原因之一：它们为你管理背压，暂停和恢复流，以使内存保持有界，无论大小不匹配如何。

## 使用 write() 的返回值手动处理

```js
// write() returns FALSE when the internal buffer is full — that's the backpressure signal
readable.on("data", (chunk) => {
  const ok = writable.write(chunk);
  if (!ok) {
    readable.pause();                       // stop reading — destination is full
    writable.once("drain", () => readable.resume()); // resume when it drains
  }
});
```

如果你手动处理流，规则是：当 `writable.write()` 返回 `false` 时，**暂停**可读流，并在可写流的 `drain` 事件上**恢复**它。忽略 `false` 返回是经典的背压 bug。

## 使用异步迭代（现代、简洁）

```js
// for-await respects backpressure naturally — it awaits each chunk's processing
for await (const chunk of readable) {
  await processSlowly(chunk); // only pulls the next chunk when ready
}
```

## 优先使用 pipeline 以确保安全

```js
import { pipeline } from "stream/promises";
await pipeline(source, transform, destination); // backpressure + error handling + cleanup
```

## 为什么这很重要

背压是任何数据密集型 Node 服务（文件处理、代理、流传输、ETL）中一个微妙但至关重要的概念。

忽视它——通过缓存所有内容或忽略 `write()` 的返回值——会导致无界的内存增长和在负载下崩溃，通常只在生产环境中处理大型数据时才会发生。

理解 `pipe`/`pipeline` 自动处理它（优先使用它们的主要原因），以及如何手动尊重它（在 `false` 时暂停，在 `drain` 时恢复）或通过 `for await`，是让你能够构建内存稳定的流系统的关键，无论数据有多大或多快。

这是胜任流使用的标志。