Node.js 流(Stream)与背压:处理大数据的正确姿势

如果你写过"把上传的大文件存到磁盘""把数据库导出成 CSV 返回给前端"这类需求,大概率用过流(Stream)。但流真正的精髓 —— 背压(backpressure) —— 很多人用了很久都没意识到。这篇就把流讲透。

为什么需要流

先看一段"反面教材":

import { readFile } from 'node:fs/promises'

app.get('/download', async (req, res) => {
  // 把整个 2GB 文件读进内存再发送
  const data = await readFile('/data/huge.bin')
  res.end(data)
})

问题很明显:文件多大,内存就涨多少。几个并发请求就能把进程内存打爆(OOM)。

流的思路完全不同:把数据切成一小块一小块(chunk),读一块、发一块、丢一块,内存占用始终维持在一个很小的水位线上。

import { createReadStream } from 'node:fs'

app.get('/download', (req, res) => {
  // 边读边发,内存里同时只存几十 KB
  createReadStream('/data/huge.bin').pipe(res)
})

四种流

Node.js 里的流分四类,记住它们的"方向"就不会混:

类型	作用	典型例子
Readable	可读,数据从这里流出	`fs.createReadStream`、HTTP 请求体
Writable	可写,数据往这里流入	`fs.createWriteStream`、HTTP 响应
Duplex	双工,可读又可写,两端独立	TCP socket
Transform	转换,读入一种数据、写出另一种	`zlib.createGzip`、加密

Transform 是最有意思的一种 —— 它既是可写流(吃数据)又是可读流(吐数据),中间做转换。压缩、加密、编码转换都靠它。

两种读取模式

Readable 流有两种工作模式,新手最容易在这踩坑:

流动模式(flowing):监听 data 事件,数据会被自动、尽快地推给你:

const rs = createReadStream('big.log')
rs.on('data', (chunk) => {
  console.log('收到', chunk.length, '字节')
})
rs.on('end', () => console.log('读完了'))

暂停模式(paused):主动调用 read() 拉取,或者用 for await...of 异步迭代(推荐,写法最清爽):

const rs = createReadStream('big.log')
for await (const chunk of rs) {
  console.log('收到', chunk.length, '字节')
}
console.log('读完了')

一旦你给流加上 data 监听器或调用 pipe,它就切到流动模式开始喷数据。

背压:流真正的灵魂

现在来到最关键的概念。设想一个场景:你从一个快的源头(SSD,读 500MB/s)往一个慢的目的地(网络,发 10MB/s)倒数据。

如果不加控制,读出来的数据会在内存里疯狂堆积,因为下游根本来不及处理。这就是需要背压的原因。

writable.write(chunk) 有个返回值,这是背压机制的核心信号:

返回 true:内部缓冲区还没满,可以继续写
返回 false:缓冲区已超过 highWaterMark(默认 16KB),你应该停下来,等 drain 事件再继续

手动实现背压是这样的(理解原理用,实际别这么写):

function copyWithBackpressure(src, dest) {
  src.on('data', (chunk) => {
    const ok = dest.write(chunk)
    if (!ok) {
      // 下游满了,暂停读取
      src.pause()
      // 等下游消化完,再恢复读取
      dest.once('drain', () => src.resume())
    }
  })
  src.on('end', () => dest.end())
}

好消息是:pipe 和 pipeline 会自动帮你处理背压。所以现实中你几乎不用手写上面那段。

pipe 的隐患与 pipeline 的救赎

pipe 写起来很爽:

createReadStream('in.txt')
  .pipe(createGzip())
  .pipe(createWriteStream('out.txt.gz'))

但它有个致命缺陷:错误不会自动向下传递,出错时也不会自动销毁前面的流,容易造成文件句柄泄露。下面这种写法,如果 createGzip 出错,in.txt 的读取流不会被关闭:

// 有隐患:每个环节都得自己 .on('error')
a.pipe(b).pipe(c)

Node 10 引入的 pipeline 解决了这个问题:它会统一处理错误、保证所有流被正确销毁,并在结束时给一个回调:

import { pipeline } from 'node:stream/promises'
import { createReadStream, createWriteStream } from 'node:fs'
import { createGzip } from 'node:zlib'

try {
  await pipeline(
    createReadStream('in.txt'),
    createGzip(),
    createWriteStream('out.txt.gz'),
  )
  console.log('压缩完成')
} catch (err) {
  console.error('管道出错,所有流已自动清理:', err)
}

经验法则:能用 pipeline 就别用 pipe,尤其是生产代码。

自己动手写一个 Transform 流

假设要做一个把每行文本转成大写的转换流:

import { Transform } from 'node:stream'

const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    // 处理这一块数据,然后把结果 push 出去
    this.push(chunk.toString().toUpperCase())
    // 通知流:这一块处理完了
    callback()
  },
})

process.stdin.pipe(upperCase).pipe(process.stdout)

transform 函数的三个要点:

chunk:进来的数据块
this.push(data):把转换结果交给下游
callback():必须调用,告诉流可以处理下一块了(出错就 callback(err))

小结

流的核心价值是用恒定的小内存处理任意大的数据,告别"全读进内存"
四种流记方向:Readable 流出、Writable 流入、Duplex 双向、Transform 边读边转
背压是流的灵魂:write() 返回 false 就该停下等 drain,否则内存会爆
生产代码优先用 pipeline,它能自动处理错误和资源清理,pipe 不行
自定义转换逻辑写 Transform,记得 push 结果并调用 callback

把背压这件事想明白,你就真正掌握了 Node.js 处理 I/O 密集场景的看家本领。