Featured image of post Node.js Streams：数据处理实用指南Featured image of post Node.js Streams：数据处理实用指南
Web

Node.js Streams：数据处理实用指南

用于高效数据处理的 Node.js 流实用指南。涵盖可读、可写、转换流、背压、管道、对象模式和内存优化。

  1. 首页
  2. Webs
  3. Node.js Streams：数据处理实用指南

Node.js 流是该平台最强大但未得到充分利用的功能之一。它们可以在数据到达时逐段处理数据，而不是将整个数据集加载到内存中。这使得它们对于处理大文件、网络通信和实时数据转换至关重要。本指南涵盖了构建强大数据管道的流基础知识和实用模式。

了解流类型

Node.js 提供四种基本流类型，每种类型在数据处理管道中发挥不同的作用。

类型方向目的示例
可读来源生产数据fs.createReadStream，HTTP 请求
可写水槽消耗数据fs.createWriteStream，HTTP 响应
转变两者修改数据zlib.createGzip、crypto.createCipher
复式两者独立读/写TCP 套接字

每个流都扩展 EventEmitter 并实现特定的内部方法。可读流使用 _read() 来获取数据，可写流使用 _write() 来接受数据，转换流使用 _transform() 来处理每个块。

管道模式

pipeline() 函数是在 Node.js 中连接流的推荐方法。与旧的 .pipe() 方法不同，pipeline() 在链中的流失败时正确传播错误并清理资源。

const { pipeline } = require("stream/promises");
const { createReadStream, createWriteStream } = require("fs");
const { createGzip } = require("zlib");

async function compressFile(source, destination) {
  try {
    await pipeline(
      createReadStream(source),
      createGzip(),
      createWriteStream(destination)
    );
    console.log("Compression complete");
  } catch (err) {
    console.error("Pipeline failed:", err);
  }
}

使用 pipeline() 时，如果读取流失败，所有下游流将自动销毁并传播错误。 .pipe() 方法缺乏这种行为 - 来自中间流的错误可能会被忽视，从而留下悬空资源。

背压：流量控制

当可读流生成数据的速度快于可写流消耗数据的速度时，就会出现背压。 Node.js 通过内部缓冲和 highWaterMark 阈值自动处理这个问题。当可写的内部缓冲区超过 highWaterMark 时，readable.pipe() 会暂停源流，直到缓冲区耗尽。

const { createReadStream, createWriteStream } = require("fs");

const readable = createReadStream("large-file.txt", { highWaterMark: 16 * 1024 });
const writable = createWriteStream("output.txt", { highWaterMark: 16 * 1024 });

writable.on("drain", () => {
  console.log("Buffer drained, resuming reading");
});

readable.on("data", (chunk) => {
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    console.log("Backpressure detected, pausing");
    readable.pause();
  }
});

drain 事件表示可写流已准备好接收更多数据。忽略背压会导致内存无限增长，并最终在高负载下导致进程崩溃。 pipeline() 函数自动处理这种协调。

对象模式流

默认情况下，流对缓冲区和字符串进行操作。对象模式流与 JavaScript 对象一起使用，这对于 JSON 转换和 CSV 解析管道非常有用。

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

const jsonParser = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    // chunk is a JavaScript object
    const transformed = {
      ...chunk,
      processedAt: new Date().toISOString(),
    };
    callback(null, transformed);
  },
});

// Usage: read NDJSON, transform objects, write NDJSON
await pipeline(
  createReadStream("input.ndjson"),
  new LineSplitter(),          // custom: NDJSON lines → objects
  jsonParser,                   // transform objects
  new ObjectStringifier(),      // custom: objects → NDJSON lines
  createWriteStream("output.ndjson")
);

由于对象分配和垃圾收集，对象模式与二进制模式相比具有性能开销。仅当您的处理逻辑需要结构化数据访问时才使用它。

使用流进行异步迭代

现代 Node.js 在可读流上支持 for await...of ，为基于事件的消费提供了更干净的替代方案：

const { createReadStream } = require("fs");
const { createInterface } = require("readline");

async function processCSV(filePath) {
  const rl = createInterface({
    input: createReadStream(filePath),
    crlfDelay: Infinity,
  });

  for await (const line of rl) {
    const row = line.split(",");
    // Process each row without loading the entire file
    await processRow(row);
  }
}

当您需要将流处理与异步/等待控制流（例如循环内的数据库插入或 HTTP 请求）集成时，此模式特别有用。

文件处理管道

流擅长处理超出可用内存的大文件。 gzip 解压缩管道演示了该模式：

const { createReadStream, createWriteStream } = require("fs");
const { createGunzip } = require("zlib");
const { pipeline } = require("stream/promises");

async function decompressLogs() {
  await pipeline(
    createReadStream("server.log.gz"),
    createGunzip(),
    createWriteStream("server.log")
  );
}

相同的模式适用于加密（使用 crypto.createDecipher）、图像处理（链接 sharp 转换）和日志解析（使用 readline 和 createInterface）。无论文件大小如何，基于流的处理都会保持内存使用量恒定，而缓冲方法则需要与输入成比例的 O(n) 内存。

错误处理和自定义流

.pipe() 和 pipeline() 之间的错误处理存在显着差异。始终将 pipeline() 用于生产代码。对于自定义流，覆盖 _transform() 并通过将 Error 传递给回调来发出错误信号：

class ValidateTransform extends Transform {
  _transform(chunk, encoding, callback) {
    if (!this._isValid(chunk)) {
      callback(new Error("Invalid data received"));
      return;
    }
    callback(null, this._process(chunk));
  }
}

finished() 实用程序检测流何时结束，无论是成功还是由于错误：

const { finished } = require("stream/promises");

try {
  await finished(readableStream);
  console.log("Stream completed");
} catch (err) {
  console.error("Stream failed:", err);
}

性能优化

调整 highWaterMark 会显着影响吞吐量。较大的值会减少读取操作的数量，但会增加内存使用量。对于高吞吐量文件传输，64 KB 到 1 MB 之间的值很常见。使用 cork() 和 uncork() 批量写入可写流，减少系统调用开销：

writable.cork();
for (const chunk of chunks) {
  writable.write(chunk);
}
process.nextTick(() => writable.uncork());

结论

Node.js 流提供了一种内存高效、可组合的数据处理方法。通过了解四种流类型，使用 pipeline() 进行错误安全连接，并考虑背压，您可以构建强大的数据管道来处理任何大小的文件。首先将大量缓冲区的操作迁移到基于流的替代方案，并测量应用程序中的内存节省情况。

©2022 - 2026 Commentary of Takao