Stream的高性能应用

Stream的概念与优势

Node.js中的Stream是处理流式数据的抽象接口，它允许数据分块处理而非一次性加载到内存。这种机制特别适合处理大文件、网络通信或实时数据流。Stream的核心优势在于内存效率，它通过事件驱动的方式逐步处理数据，避免内存爆仓。

const fs = require('fs');

// 传统文件读取方式
fs.readFile('largefile.txt', (err, data) => {
  // 整个文件内容会加载到内存
});

// Stream方式读取
const readStream = fs.createReadStream('largefile.txt');
readStream.on('data', (chunk) => {
  // 每次只处理一小块数据
});

Stream的四种基本类型

Node.js提供了四种基础Stream类型，每种类型对应不同的数据处理场景：

1. Readable Stream：数据源流，如文件读取、HTTP请求
2. Writable Stream：数据目标流，如文件写入、HTTP响应
3. Duplex Stream：双向流，如TCP socket
4. Transform Stream：数据处理流，如压缩/解压

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

// 自定义Transform流示例
const upperCaseTr = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

process.stdin.pipe(upperCaseTr).pipe(process.stdout);

高性能流处理技巧

背压控制机制

当数据生产速度超过消费速度时，Stream会自动启动背压机制。理解这个机制对构建高性能应用至关重要：

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

readable.on('data', (chunk) => {
  // 检查写入流是否处于背压状态
  if (!writable.write(chunk)) {
    readable.pause();  // 暂停读取
  }
});

writable.on('drain', () => {
  readable.resume();  // 恢复读取
});

管道连接优化

使用pipe()方法可以自动处理背压，但多级管道连接时需要注意：

const zlib = require('zlib');

// 文件压缩管道
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'))
  .on('finish', () => console.log('压缩完成'));

实际应用场景

大文件处理

处理GB级日志文件时，Stream可以保持内存使用稳定：

const readline = require('readline');
const rl = readline.createInterface({
  input: fs.createReadStream('huge.log'),
  crlfDelay: Infinity
});

let errorCount = 0;
rl.on('line', (line) => {
  if (line.includes('ERROR')) errorCount++;
}).on('close', () => {
  console.log(`总错误数: ${errorCount}`);
});

实时视频转码

使用Stream构建视频处理流水线：

const ffmpeg = require('fluent-ffmpeg');

function transcodeVideo(inputPath, outputPath) {
  return new Promise((resolve, reject) => {
    ffmpeg(inputPath)
      .videoCodec('libx264')
      .audioCodec('aac')
      .output(outputPath)
      .on('end', resolve)
      .on('error', reject)
      .run();
  });
}

性能调优实践

缓冲区大小调整

根据应用场景调整缓冲区大小可以显著提升性能：

// 调整读取流缓冲区大小
const readStream = fs.createReadStream('large.dat', {
  highWaterMark: 1024 * 1024 * 8 // 8MB
});

// 调整写入流缓冲区大小
const writeStream = fs.createWriteStream('output.dat', {
  highWaterMark: 1024 * 1024 * 4 // 4MB
});

并行流处理

利用Worker Threads实现流处理的并行化：

const { Worker } = require('worker_threads');

function processInWorker(stream) {
  return new Promise((resolve) => {
    const worker = new Worker('./stream-worker.js', {
      workerData: { stream }
    });
    worker.on('message', resolve);
  });
}

错误处理与调试

健壮的错误处理机制

Stream的错误处理需要特别注意，因为错误可能发生在管道的任何环节：

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

性能监控与调试

使用stream.pipeline和性能钩子监控流性能：

const { performance, PerformanceObserver } = require('perf_hooks');

const obs = new PerformanceObserver((items) => {
  items.getEntries().forEach((entry) => {
    console.log(`${entry.name}: ${entry.duration}ms`);
  });
});
obs.observe({ entryTypes: ['measure'] });

performance.mark('stream-start');
const stream = fs.createReadStream('data.bin')
  .on('end', () => {
    performance.mark('stream-end');
    performance.measure('Stream Duration', 'stream-start', 'stream-end');
  });

高级应用模式

自定义协议实现

基于Stream实现自定义网络协议：

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

class MessageParser extends Transform {
  constructor() {
    super({ readableObjectMode: true });
    this.buffer = Buffer.alloc(0);
  }

  _transform(chunk, encoding, callback) {
    this.buffer = Buffer.concat([this.buffer, chunk]);
    while (this.buffer.length >= 4) {
      const length = this.buffer.readUInt32BE(0);
      if (this.buffer.length >= 4 + length) {
        const message = this.buffer.slice(4, 4 + length);
        this.push(message.toString());
        this.buffer = this.buffer.slice(4 + length);
      } else {
        break;
      }
    }
    callback();
  }
}

const server = net.createServer((socket) => {
  socket.pipe(new MessageParser()).on('data', console.log);
});

流式数据库操作

处理大规模数据库导出/导入：

const { Client } = require('pg');
const client = new Client();

async function exportLargeTable() {
  await client.connect();
  const stream = client.query(new Query('SELECT * FROM huge_table')).stream();
  
  const transform = new Transform({
    objectMode: true,
    transform(row, encoding, callback) {
      this.push(JSON.stringify(row) + '\n');
      callback();
    }
  });

  stream.pipe(transform).pipe(fs.createWriteStream('export.jsonl'));
}