深入 Node.js Stream 流：处理大数据的利器

深入 Node.js Stream 流：处理大数据的利器

Stream 是 Node.js 最强大但最被低估的特性之一。它让你能够以内存高效的方式处理大量数据。

为什么需要 Stream？

// ❌ 传统方式：将整个文件读入内存
const data = fs.readFileSync('huge-file.csv');  // 可能 OOM！
processData(data);

// ✅ Stream 方式：逐块处理
fs.createReadStream('huge-file.csv')
  .pipe(csvParser())
  .pipe(transformStream())
  .pipe(fs.createWriteStream('output.csv'));

四种 Stream 类型

类型	功能	示例
Readable	读取数据	fs.createReadStream()
Writable	写入数据	fs.createWriteStream()
Duplex	读写	net.Socket
Transform	转换数据	zlib.createGzip()

Readable Stream

const { Readable } = require('stream');

// 自定义可读流
class NumberGenerator extends Readable {
  constructor(max, options) {
    super(options);
    this.max = max;
    this.current = 0;
  }

  _read() {
    if (this.current < this.max) {
      this.push(String(this.current++));
    } else {
      this.push(null); // 结束
    }
  }
}

const numbers = new NumberGenerator(1000000);

// 两种消费方式
// 方式 1：pipe
numbers.pipe(process.stdout);

// 方式 2：事件
numbers.on('data', chunk => console.log(chunk));
numbers.on('end', () => console.log('Done'));

Transform Stream

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

class UppercaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

// 使用
fs.createReadStream('input.txt')
  .pipe(new UppercaseTransform())
  .pipe(fs.createWriteStream('output.txt'));

Pipeline

const { pipeline } = require('stream/promises');

async function processFile() {
  await pipeline(
    fs.createReadStream('input.csv'),
    csvParser(),
    filterStream(row => row.active),
    transformStream(row => ({ ...row, processed: true })),
    stringify(),
    fs.createWriteStream('output.csv')
  );
  console.log('Pipeline completed');
}

背压处理

当写入速度跟不上读取速度时，Stream 自动处理背压：

const readable = getFastReadable();
const writable = getSlowWritable();

// pipe 自动处理背压
readable.pipe(writable);

// 手动处理背压
readable.on('data', chunk => {
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    readable.pause(); // 暂停读取
    writable.once('drain', () => readable.resume()); // 等待排空
  }
});

实战：处理 10GB CSV 文件

async function processLargeCSV(inputPath, outputPath) {
  const lineCount = { value: 0 };
  
  await pipeline(
    fs.createReadStream(inputPath),
    split2(),                          // 按行分割
    new Transform({
      objectMode: true,
      transform(line, enc, cb) {
        lineCount.value++;
        if (lineCount.value % 100000 === 0) {
          console.log(`Processed ${lineCount.value} lines`);
        }
        const processed = processLine(line.toString());
        cb(null, processed + '\n');
      }
    }),
    fs.createWriteStream(outputPath)
  );
}

总结

Stream 的核心价值：

• 内存高效：逐块处理，不会 OOM
• 时间高效：边读边处理，无需等待全部读完
• 可组合：通过 pipe 和 pipeline 连接
• 自动背压：防止内存溢出