Node.js 流式处理实战：Readable、Writable、Transform 与背压机制全解析

为什么用流？内存从 GB 降到 KB

方式	处理 1GB 文件	内存占用	耗时
fs.readFile	全量读入内存	1GB+	长且阻塞
fs.createReadStream	分块处理	~64KB	立即开始

核心思想：不等待全部数据就绪，边读边处理——数据像水流一样经过管道。

---

一、四种流类型

类型	方向	典型用途
Readable	输入	文件读取、HTTP 请求体
Writable	输出	文件写入、HTTP 响应
Duplex	双向	TCP Socket、WebSocket
Transform	转换	压缩、加密、格式转换

继承关系

Stream
├── Readable
├── Writable
├── Duplex (Readable + Writable)
└── Transform (Duplex，输出基于输入)

---

二、Readable 流

两种模式

模式	触发方式	特点
Paused（暂停）	默认	需手动调用 read()
Flowing（流动）	绑定 data 事件	自动推送数据

创建自定义 Readable

import { Readable } from 'stream';

class NumberStream extends Readable {
  constructor(max) {
    super({ objectMode: true });
    this.max = max;
    this.current = 0;
  }

  _read() {
    if (this.current < this.max) {
      this.push({ value: this.current, timestamp: Date.now() });
      this.current++;
    } else {
      this.push(null); // 结束流
    }
  }
}

const stream = new NumberStream(5);
stream.on('data', (chunk) => console.log(chunk));
// { value: 0, timestamp: ... }
// { value: 1, timestamp: ... }
// ...

从迭代器创建

import { Readable } from 'stream';

async function* generateLogs() {
  for (let i = 0; i < 100; i++) {
    await sleep(100);
    yield `[${new Date().toISOString()}] Event ${i}\n`;
  }
}

const logStream = Readable.from(generateLogs());
logStream.pipe(process.stdout);

---

三、Writable 流

创建自定义 Writable

import { Writable } from 'stream';

class BatchWriter extends Writable {
  constructor(options) {
    super({ objectMode: true, highWaterMark: 10 });
    this.batch = [];
    this.batchSize = options.batchSize || 5;
  }

  _write(chunk, encoding, callback) {
    this.batch.push(chunk);

    if (this.batch.length >= this.batchSize) {
      this.flush()
        .then(() => callback())
        .catch(callback);
    } else {
      callback();
    }
  }

  _final(callback) {
    if (this.batch.length > 0) {
      this.flush()
        .then(() => callback())
        .catch(callback);
    } else {
      callback();
    }
  }

  async flush() {
    console.log('写入批次:', this.batch);
    await db.batchInsert(this.batch);
    this.batch = [];
  }
}

write() 返回值与背压

const writable = fs.createWriteStream('output.txt');

for (let i = 0; i < 1e6; i++) {
  const canContinue = writable.write(`Line ${i}\n`);
  if (!canContinue) {
    // 缓冲区已满，等待 drain 事件
    await once(writable, 'drain');
  }
}
writable.end();

---

四、Transform 流

基础 Transform

import { Transform } from 'stream';

class JsonLineParser extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  _transform(chunk, encoding, callback) {
    try {
      const data = JSON.parse(chunk.toString());
      this.push(data);
      callback();
    } catch (err) {
      callback(err);
    }
  }
}

实战：CSV 转 JSON 流

import { Transform } from 'stream';

class CsvToJson extends Transform {
  constructor() {
    super({ writableObjectMode: false, readableObjectMode: true });
    this.headers = null;
    this.partialLine = '';
  }

  _transform(chunk, encoding, callback) {
    const lines = (this.partialLine + chunk.toString()).split('\n');
    this.partialLine = lines.pop(); // 保留不完整的行

    for (const line of lines) {
      if (!this.headers) {
        this.headers = line.split(',');
        continue;
      }

      const values = line.split(',');
      const obj = {};
      this.headers.forEach((h, i) => obj[h.trim()] = values[i]?.trim());
      this.push(obj);
    }

    callback();
  }

  _flush(callback) {
    if (this.partialLine) {
      const values = this.partialLine.split(',');
      const obj = {};
      this.headers.forEach((h, i) => obj[h.trim()] = values[i]?.trim());
      this.push(obj);
    }
    callback();
  }
}

使用 pipeline 组合

import { pipeline } from 'stream/promises';
import { createReadStream, createWriteStream } from 'fs';
import { CsvToJson } from './csv-to-json.js';
import { Transform } from 'stream';

await pipeline(
  createReadStream('data.csv'),
  new CsvToJson(),
  new Transform({
    objectMode: true,
    transform(obj, _, cb) {
      this.push(JSON.stringify(obj) + '\n');
      cb();
    },
  }),
  createWriteStream('data.jsonl')
);

pipeline 的优势：自动处理错误传播和流清理，比 .pipe() 更安全。

---

五、背压（Backpressure）机制

什么是背压？

当生产者速度 > 消费者速度时，数据在缓冲区堆积。背压是消费者通知生产者减速的机制。

Readable (快) → Writable (慢)
  产生数据 100MB/s    只能处理 10MB/s
  
  无背压：缓冲区爆满 → 内存溢出
  有背压：Readable 暂停 → 等待 Writable 消化

背压的工作流程

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    readable.pause(); // 暂停读取
    writable.once('drain', () => {
      readable.resume(); // 恢复读取
    });
  }
});

highWaterMark 配置

流类型	默认值	含义
Readable	64KB (16KB objectMode)	内部缓冲区大小
Writable	16KB	写入缓冲区大小

const readable = fs.createReadStream('big.txt', {
  highWaterMark: 1024 * 1024, // 1MB 缓冲区
});

---

六、实战案例

案例1：大文件 Gzip 压缩

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

await pipeline(
  createReadStream('large.log'),
  createGzip({ level: 9 }),
  createWriteStream('large.log.gz')
);

案例2：HTTP 流式响应

import { createReadStream } from 'fs';
import { stat } from 'fs/promises';

app.get('/video/:id', async (req, res) => {
  const filePath = `/videos/${req.params.id}.mp4`;
  const { size } = await stat(filePath);
  const range = req.headers.range;

  if (range) {
    const [start, end] = range.replace(/bytes=/, '').split('-').map(Number);
    res.writeHead(206, {
      'Content-Range': `bytes ${start}-${end || size - 1}/${size}`,
      'Content-Length': (end || size - 1) - start + 1,
      'Content-Type': 'video/mp4',
    });
    createReadStream(filePath, { start, end }).pipe(res);
  } else {
    res.writeHead(200, {
      'Content-Length': size,
      'Content-Type': 'video/mp4',
    });
    createReadStream(filePath).pipe(res);
  }
});

案例3：逐行处理日志

import { pipeline } from 'stream/promises';
import { createReadStream } from 'fs';
import { Transform } from 'stream';
import { createInterface } from 'readline';

const fileStream = createReadStream('app.log');
const rl = createInterface({ input: fileStream });

let errorCount = 0;
for await (const line of rl) {
  if (line.includes('ERROR')) {
    errorCount++;
    if (errorCount <= 10) console.log(line);
  }
}
console.log(`总错误数: ${errorCount}`);

案例4：Web Stream 与 Node Stream 互转

import { Readable } from 'stream';

// Node Readable → Web ReadableStream
const nodeStream = fs.createReadStream('data.bin');
const webStream = Readable.toWeb(nodeStream);

// 在浏览器端使用
const reader = webStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  processChunk(value);
}

---

七、常见陷阱

陷阱	后果	解决
忘记处理 error 事件	进程静默崩溃	用 pipeline 代替 .pipe()
.pipe() 不处理背压	内存溢出	用 pipeline 或手动处理
Transform 中不调用 callback	流挂起	确保 _transform 中调用 callback
objectMode 混用	数据丢失/类型错误	确保上下游 objectMode 一致
不调用 _final	最后一批数据丢失	实现 _final 处理残余数据