Node.js Streams：高效处理大数据流

什么是 Stream？

Stream 是 Node.js 处理流式数据的强大工具，允许你逐块处理数据而不是一次性加载到内存。

四种 Stream 类型

Readable - 可读流

const fs = require("fs");
const readStream = fs.createReadStream("large-file.txt", {
  highWaterMark: 16 * 1024  // 16KB chunks
});

readStream.on("data", (chunk) => {
  console.log("Received chunk:", chunk.length);
});

readStream.on("end", () => {
  console.log("Stream ended");
});

Writable - 可写流

const writeStream = fs.createWriteStream("output.txt");

for (let i = 0; i < 1000; i++) {
  writeStream.write(`Line ${i}\n`);
}

writeStream.end();

Duplex - 双工流

既可读又可写，如 net.Socket。

Transform - 转换流

可修改数据，如 zlib.createGzip()。

Pipe - 流式管道

const fs = require("fs");
const zlib = require("zlib");

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

实战：处理大文件

const fs = require("fs");
const readline = require("readline");

async function processLargeFile(filePath) {
  const fileStream = fs.createReadStream(filePath);
  const rl = readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity
  });

  let lineCount = 0;
  for await (const line of rl) {
    lineCount++;
    // 处理每一行
  }
  console.log(`Total lines: ${lineCount}`);
}

背压处理

const readStream = fs.createReadStream("input.txt");
const writeStream = fs.createWriteStream("output.txt");

readStream.on("data", (chunk) => {
  const canContinue = writeStream.write(chunk);
  if (!canContinue) {
    readStream.pause();
  }
});

writeStream.on("drain", () => {
  readStream.resume();
});

优势

• 内存高效: 不一次性加载整个文件
• 时间高效: 边读边处理
• 可组合: 通过 pipe 链接多个操作

Stream 是 Node.js 处理 I/O 的核心机制，掌握它能让你构建高性能应用。