Node.js 中的流（Stream）及其应用场景

在 Node.js 中，流（Stream）是一种处理数据的强大工具，尤其是在处理大量数据或需要逐块处理数据的场景中。流的核心思想是将数据分块处理，而不是一次性加载到内存中，从而提高了性能和资源利用率。本文将介绍 Node.js 中流的类型、工作原理以及常见的应用场景。

什么是流？

流是 Node.js 中处理流式数据的抽象接口。它们允许我们以逐块的方式读取或写入数据，而不需要一次性将所有数据加载到内存中。流特别适合处理以下场景：

大文件读写
实时数据传输
数据管道操作

Node.js 中的流分为四种类型：

可读流（Readable Stream）：用于读取数据（例如从文件中读取数据）。
可写流（Writable Stream）：用于写入数据（例如向文件中写入数据）。
双工流（Duplex Stream）：既可读又可写（例如网络套接字）。
转换流（Transform Stream）：在读写过程中可以修改或转换数据（例如压缩或加密数据）。

流的工作原理

流的核心思想是通过事件驱动的方式处理数据。以下是流的基本工作流程：

数据分块：流将数据分成小块（chunks），逐块处理。
事件驱动：流通过事件（如 data、end、error）通知应用程序数据的可用性。
管道操作：多个流可以通过管道（pipe）连接起来，形成一个数据处理链。

流的应用场景

1. 大文件读写

当处理大文件时，一次性将文件加载到内存中可能会导致内存溢出。使用流可以逐块读取或写入文件，从而避免内存问题。

示例：使用流复制大文件

const fs = require("fs");

// 创建可读流和可写流
const readableStream = fs.createReadStream("largefile.txt");
const writableStream = fs.createWriteStream("copy-largefile.txt");

// 通过管道将数据从可读流传输到可写流
readableStream.pipe(writableStream);

console.log("文件复制完成！");

2. 实时数据传输

流非常适合处理实时数据，例如视频流、日志流或传感器数据。通过流，数据可以在生成的同时被处理，而不需要等待所有数据都准备好。

const fs = require("fs");
// 创建一个可读流来读取日志文件
const logStream = fs.createReadStream("app.log", "utf8");
// 监听数据事件，逐块处理日志
logStream.on("data", (chunk) => {
  console.log("收到日志数据：", chunk);
});
logStream.on("end", () => {
  console.log("日志读取完成！");
});

3. 数据压缩与解压缩

使用转换流可以对数据进行压缩或解压缩。Node.js 提供了 zlib 模块来支持常见的压缩算法。

const fs = require("fs");
const zlib = require("zlib");
const path = require("path");
class FileCompressor {
  constructor(inputPath, outputPath) {
    this.inputPath = inputPath;
    this.outputPath = outputPath;
  }
  compress() {
    return new Promise((resolve, reject) => {
      // 创建流
      const readStream = fs.createReadStream(this.inputPath);
      const writeStream = fs.createWritStream(this.outputPath);
      const gzip = zlib.createGzip();
      // 监控压缩进度
      let inputSize = 0;
      let outputSize = 0;
      readStream.on("data", (chunk) => {
        inputSize += chunk.length;
      });
      gzip.on("data", (chunk) => {
        outputSize += chunk.length;
        const ratio = ((1 - outputSize / inputSize) * 100).toFixed(2);
        console.log(`压缩率: ${ratio}%`);
      });
      // 链接流
      readStream
        .pipe(gzip)
        .pipe(writeStream)
        .on("finish", () => {
          console.log("文件压缩完成！");
          console.log(`原始大小: ${inputSize} bytes`);
          console.log(`压缩后大小: ${outputSize} bytes`);
          resolve();
        })
        .on("error", (error) => {
          console.error("压缩过程出错:", error);
          reject(error);
        });
    });
  }
}
// 使用示例
async function compressFile() {
  const compressor = new FileCompressor(
    path.join(__dirname, "input.txt"),
    path.join(__dirname, "output.txt.gz")
  );
  try {
    await compressor.compress();
    console.log("压缩操作成功完成");
  } catch (error) {
    console.error("压缩操作失败:", error);
  }
}
compressFile();

4. HTTP 流式响应

在 Node.js 的 HTTP 模块中，请求和响应对象都是流。这使得我们可以高效地处理大型文件上传或下载。

const http = require("http");
const fs = require("fs");
const server = http.createServer((req, res) => {
  const fileStream = fs.createReadStream("largefile.zip");
  res.writeHead(200, { "Content-Type": "application/zip" });
  fileStream.pipe(res); // 将文件流直接传输到响应中
});
server.listen(3000, () => {
  console.log("服务器正在运行：http://localhost:3000");
});

流的高级应用

数据转换流不仅可以用于数据传输，还可以用于数据转换。例如，你可以创建一个转换流来处理 CSV 文件：

const { Transform } = require("stream");
class CSVTransform extends Transform {
  constructor(options = {}) {
    super(options);
    this.headers = null;
  }
  _transform(chunk, encoding, callback) {
    const data = chunk.toString();
    const rows = data.split("\n");
    if (!this.headers) {
      this.headers = rows[0].split(",");
      rows.shift();
    }
    const jsonData = rows.map((row) => {
      const values = row.split(",");
      return this.headers.reduce((obj, header, index) => {
        obj[header] = values[index];
        return obj;
      }, {});
    });
    this.push(JSON.stringify(jsonData));
    callback();
  }
}

内存使用优化流的一个重要应用是优化内存使用。比如处理大型日志文件时：

const readline = require("readline");
const fs = require("fs");
const rl = readline.createInterface({
  input: fs.createReadStream("large_log.txt"),
  crlfDelay: Infinity,
});
rl.on("line", (line) => {
  // 逐行处理日志
  if (line.includes("ERROR")) {
    console.log(`发现错误: ${line}`);
  }
});

总结

Node.js 中的流是一种强大的工具，特别适合处理大文件、实时数据以及需要逐块处理的场景。通过流，我们可以高效地管理内存资源，并实现复杂的数据处理逻辑。无论是文件操作、网络通信还是数据转换，流都能提供优雅的解决方案。

Node.js中的流应用场景