目录

Node.js流式处理实战

Stream 是 Node.js 处理大数据的核心机制，能够高效处理 GB 级数据而不会内存溢出。

流的类型

类型 说明 示例
Readable 可读流 fs.createReadStream
Writable 可写流 fs.createWriteStream
Duplex 双工流 TCP socket
Transform 转换流 zlib.createGzip

基本用法

读取大文件

javascript
const fs = require('fs');

// 传统方式：一次性读取，大文件会爆内存
// const data = fs.readFileSync('big.txt');

// 流式读取：分块处理
const readStream = fs.createReadStream('big.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024  // 64KB 缓冲区
});

readStream.on('data', (chunk) => {
  console.log(`收到 ${chunk.length} 字节`);
});

readStream.on('end', () => {
  console.log('读取完成');
});

pipe 管道

javascript
// 文件复制：自动处理背压
fs.createReadStream('input.txt')
  .pipe(fs.createWriteStream('output.txt'));

// 压缩文件
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('output.txt.gz'));

自定义 Transform 流

javascript
const { Transform } = require('stream');

class JsonLineParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('
');

    this.buffer = lines.pop(); // 保留不完整的行

    for (const line of lines) {
      if (line.trim()) {
        try {
          this.push(JSON.parse(line));
        } catch (e) {
          callback(e);
          return;
        }
      }
    }
    callback();
  }

  _flush(callback) {
    if (this.buffer.trim()) {
      this.push(JSON.parse(this.buffer));
    }
    callback();
  }
}

// 处理大型 JSON Lines 文件
fs.createReadStream('data.jsonl')
  .pipe(new JsonLineParser())
  .on('data', (obj) => {
    console.log(obj);
  });

pipeline（推荐）

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

// pipeline 自动处理错误和背压
pipeline(
  fs.createReadStream('input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('output.txt.gz'),
  (err) => {
    if (err) {
      console.error('管道失败:', err);
    } else {
      console.log('管道成功');
    }
  }
);

处理 HTTP 请求流

javascript
const http = require('http');

// 流式响应
http.createServer((req, res) => {
  pipeline(
    fs.createReadStream('video.mp4'),
    res,
    (err) => {
      if (err) console.error(err);
    }
  );
}).listen(3000);

// 流式接收上传
app.post('/upload', (req, res) => {
  const writeStream = fs.createWriteStream('uploaded.file');

  req.on('data', (chunk) => {
    const ok = writeStream.write(chunk);
    if (!ok) {
      req.pause();
      writeStream.once('drain', () => req.resume());
    }
  });

  req.on('end', () => {
    writeStream.end();
    res.json({ success: true });
  });
});

背压（Backpressure）

当写入速度慢于读取速度时，需要处理背压：

javascript
// 手动处理背压
const readable = fs.createReadStream('source.txt');
const writable = fs.createWriteStream('dest.txt');

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    readable.pause();
    writable.once('drain', () => readable.resume());
  }
});

// 推荐：使用 pipe 或 pipeline 自动处理
readable.pipe(writable);

总结

Node.js Stream 是处理大数据的利器。通过分块读取、管道连接和自动背压处理，可以高效处理远超内存容量的数据。实际开发中优先使用 pipeline 而非手动 pipe，它能更好地处理错误和资源清理。