Node.js Stream 流

Stream 是 Node.js 处理大量数据的核心抽象，核心思想：把数据拆成小块，边读边处理，不占大量内存。

四种流类型

类型	说明	示例
Readable	可读流	fs.createReadStream、http.IncomingMessage
Writable	可写流	fs.createWriteStream、http.ServerResponse
Duplex	双工流（可读可写）	net.Socket、TLS
Transform	转换流（读写中间处理）	zlib.createGzip、crypto.createCipher

Readable 可读流

import { createReadStream } from 'node:fs'

const rs = createReadStream('./big-file.mp4', { highWaterMark: 64 * 1024 }) // 64KB 缓冲区

// 两种读取模式：
// 1. flowing 模式（自动推送）
rs.on('data', (chunk) => console.log('收到:', chunk.length))
rs.on('end', () => console.log('读取完毕'))
rs.on('error', (err) => console.error(err))

// 2. paused 模式（手动拉取）
rs.on('readable', () => {
  let chunk
  while ((chunk = rs.read()) !== null) {
    console.log('手动读取:', chunk.length)
  }
})

Writable 可写流

import { createWriteStream } from 'node:fs'

const ws = createWriteStream('./output.txt')

ws.write('hello\n')
ws.write('world\n')
ws.end('最后一行')  // 结束写入

ws.on('finish', () => console.log('写入完成'))
ws.on('error', (err) => console.error(err))

背压处理（防止内存溢出）：

const rs = createReadStream('./big.mp4')
const ws = createWriteStream('./copy.mp4')

rs.on('data', (chunk) => {
  const canWrite = ws.write(chunk)
  if (!canWrite) {
    rs.pause()  // 写满了，暂停读
    ws.once('drain', () => rs.resume())  // 排空了，继续读
  }
})

最简单的方式是用 pipe，自动处理背压： rs.pipe(ws)

Transform 转换流

import { Transform } from 'node:stream'

// 大写转换流
const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase())
  }
})

fs.createReadStream('./input.txt')
  .pipe(upperCase)
  .pipe(fs.createWriteStream('./output.txt'))

管道 pipeline（推荐）

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

await pipeline(
  fs.createReadStream('./input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('./output.txt.gz')
)
// 自动处理背压、错误传递、资源清理

自定义流

import { Readable } from 'node:stream'

// 自定义可读流
const customReadable = new Readable({
  read(size) {
    this.push(Buffer.from('data chunk'))
    this.push(null)  // 结束信号
  }
})

// 自定义可写流
const customWritable = new Writable({
  write(chunk, encoding, callback) {
    console.log('写入:', chunk.toString())
    callback()
  }
})

Stream events 速查

Readable 事件	Writable 事件
data — 有数据可读	drain — 缓冲区排空
end — 读完	finish — 写入完成
error — 出错	error — 出错
readable — 可手动 read	pipe — 被 pipe