Node.js 中的 Stream：处理大文件而不用尽内存

1. 总览

将整个文件读入内存使用 fs.readFile 在文件很小时非常方便。对于几 GB 的日志文件或有数百万行的 CSV 文件，这种方式会让进程的 RAM 膨胀或崩溃。本文将指导你如何在 Node.js 中使用 stream 处理大文件：逐个读取 chunk、分行、过滤数据、压缩并写入新文件。你还会理解 backpressure，以及如何用 pipeline 正确处理错误。

所有示例都使用 Node.js 20+ 和 ES module。package.json 文件需要有 "type": "module"。本文不需要外部库，因为 fs、stream、zlib 和 readline 都是内置的。

2. readFile 的问题

首先，我们创建一个约 500 万行的 orders.csv 文件来测试：

// generate-orders.mjs
import { createWriteStream } from 'node:fs';
import { once } from 'node:events';

const out = createWriteStream('orders.csv');
out.write('id,customer,amount,status\n');

for (let i = 1; i <= 5_000_000; i++) {
  const status = i % 10 === 0 ? 'FAILED' : 'PAID';
  const ok = out.write(`${i},customer-${i % 1000},${(i % 500) + 0.99},${status}\n`);
  if (!ok) await once(out, 'drain'); // 等待缓冲区被清空后继续写入
}
out.end();

这段代码已经使用了 backpressure。第 5 节将详细解释该机制。生成的文件大约 180 MB。

现在让我们用“幼稚”的方式来计算 FAILED 订单的数量：

// naive-count.mjs
import { readFile } from 'node:fs/promises';

const content = await readFile('orders.csv', 'utf8');
const failed = content.split('\n').filter(line => line.endsWith(',FAILED')).length;

console.log('FAILED:', failed);
console.log('RSS MB:', Math.round(process.memoryUsage().rss / 1024 / 1024));

FAILED: 500000
RSS MB: 1046

180 MB 的文件已经花费了大约 1 GB 的 RAM。进程必须同时保持三样东西：原始 buffer、UTF-8 字符串和 500 万个子字符串的数组。当文件大于约 2 GB 时，readFile 会抛出 ERR_FS_FILE_TOO_LARGE 错误并无法继续读取。

3. 四种基本的 stream 类型

Stream 是按块（chunk）通过的数据。Node.js 有四种 stream：

类型	作用	示例
Readable	数据源	fs.createReadStream、process.stdin
Writable	数据目标	fs.createWriteStream、process.stdout
Duplex	双向独立	TCP socket
Transform	输出取决于输入	zlib.createGzip()

每个 stream 都有一个内部 buffer。此 buffer 的限制由 highWaterMark 决定。fs.createReadStream 默认使用 64 KiB。对象模式下的 stream 使用 16 个对象。

4. 按 chunk 读取大文件

4.1. 在 Readable 上使用 for await

消费 Readable 最简洁的方式是异步迭代：

// read-chunks.mjs
import { createReadStream } from 'node:fs';

const stream = createReadStream('orders.csv', { highWaterMark: 64 * 1024 });

let bytes = 0;
let chunks = 0;
for await (const chunk of stream) {
  bytes += chunk.length;
  chunks++;
}

console.log({ chunks, mb: (bytes / 1024 / 1024).toFixed(1) });
console.log('RSS MB:', Math.round(process.memoryUsage().rss / 1024 / 1024));

{ chunks: 2853, mb: 178.3 }
RSS MB: 52

每个 chunk 是最多 64 KiB 的 Buffer。循环只在处理完当前 chunk 后才请求下一个 chunk。因此 RAM 几乎不受文件大小的影响。

4.2. 用 readline 逐行读取

Chunk 按字节切割，而不是按行。如果需要逐行处理，请使用 readline：

// count-failed.mjs
import { createReadStream } from 'node:fs';
import readline from 'node:readline';

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // 将 \r\n 视为一次换行
});

let failed = 0;
for await (const line of rl) {
  if (line.endsWith(',FAILED')) failed++;
}
console.log('FAILED:', failed);

FAILED: 500000

结果与第 2 节相同，但 RSS 仅约 50-60 MB。readline 会自动拼接两个 chunk 之间被切割的行部分，所以你不用担心这个问题。

5. 写文件和 backpressure

当数据源快于写入目标时会发生 backpressure。磁盘、网络或 gzip 都可能比读取速度慢。当内部 buffer 超过 highWaterMark 时，函数 writable.write() 返回 false。这是“停止，等待 drain 事件”的信号。

ví dụ sau bỏ qua tín hiệu này：

// bad-write.mjs — 不要这样做
import { createWriteStream } from 'node:fs';

const out = createWriteStream('big.txt');
for (let i = 0; i < 10_000_000; i++) {
  out.write(`line ${i}\n`); // 忽略返回值
}
out.end();

循环同步运行，所以所有 1000 万个字符串都会被推入 out 的 buffer 中，然后磁盘才能跟上写入。RAM 急剧增加到数 GB。修复方法是检查返回值，然后 await once(out, 'drain')，就像第 2 节中的 generate-orders.mjs 文件。

如果你用 pipeline 连接 stream，backpressure 会自动处理。这是大多数情况下应该使用的方法。

6. 使用 pipeline 连接 stream

6.1. 复制和压缩文件

// gzip-orders.mjs
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';

await pipeline(
  createReadStream('orders.csv'),
  createGzip(),
  createWriteStream('orders.csv.gz'),
);
console.log('Done');

Done

来自 node:stream/promises 的 pipeline 返回一个 Promise。当最后一个 stream 完成写入时，Promise 会 resolve。如果任何 stream 出错（例如文件不存在或磁盘已满），Promise 会 reject。此时所有 stream 都会被 destroy，所以文件描述符不会泄漏。

注意： 避免使用 a.pipe(b).pipe(c)。pipe 不会将错误传递给 stream。如果 a 出错，c 仍然会打开。请改用 pipeline。

6.2. 编写 Transform 来分行

假设你想过滤 FAILED 订单然后压缩。你需要一个可以放入 pipeline 的分行步骤。我们自己编写一个 Transform：

// split-lines.mjs
import { Transform } from 'node:stream';
import { StringDecoder } from 'node:string_decoder';

export function splitLines() {
  const decoder = new StringDecoder('utf8');
  let tail = '';

  return new Transform({
    readableObjectMode: true, // 输出端发出单个行对象，不是 Buffer

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // 最后部分可能是不完整的行
      for (const line of lines) this.push(line);
      callback();
    },

    flush(callback) {
      const rest = tail + decoder.end();
      if (rest) this.push(rest);
      callback();
    },
  });
}

逐部分解释：

• readableObjectMode: true：读取端将每一行作为单独的对象发出。写入端仍然接收 Buffer。
• StringDecoder：chunk 可能会在多字节 UTF-8 字符的中间切割，例如“Nguyễn Văn Hiệp”中的字符“ệ”。直接调用 chunk.toString() 会产生错误的字符 �。decoder.write() 会保留不完整的字节，并在下一次调用时将它们拼接在一起。
• tail：保存未完成的行部分以便与下一个 chunk 拼接。
• flush：在输入结束时运行。如果文件末尾没有 \n，此步骤会推出最后一行。

6.3. 用 async generator 过滤

pipeline 接受 async generator 作为一个步骤。因此你可以过滤而不需要编写额外的 Transform 类：

// export-failed.mjs
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';
import { splitLines } from './split-lines.mjs';

let count = 0;

await pipeline(
  createReadStream('orders.csv'),
  splitLines(),
  async function* (lines) {
    for await (const line of lines) {
      if (line.endsWith(',FAILED')) {
        count++;
        yield line + '\n';
      }
    }
  },
  createGzip(),
  createWriteStream('failed-orders.csv.gz'),
);

console.log(`Exported ${count} failed orders`);

Exported 500000 failed orders

数据流经五个步骤。每个步骤在内存中只保持几个 chunk。当 gzip 较慢时，generator 会在 yield 处阻塞，而 splitLines 停止接收 chunk。结果是 createReadStream 暂停读取磁盘。这正是 backpressure 自动运行的表现。

6.4. 用 AbortSignal 中途取消

对于长时间运行的任务，你应该允许取消或设置超时：

// export-with-timeout.mjs
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';

try {
  await pipeline(
    createReadStream('orders.csv'),
    createGzip(),
    createWriteStream('orders.csv.gz'),
    { signal: AbortSignal.timeout(500) }, // 500 ms 后取消
  );
} catch (err) {
  console.error(err.name); // 如果超时则为 AbortError
}

AbortError

被取消时，所有 stream 都会被关闭。此时文件 orders.csv.gz 只是被部分写入。在实际代码中，在 catch 块中删除该文件，或写入临时文件，完成后 rename。

7. 测试 Transform

最容易出错的逻辑是拼接两个 chunk 之间被切割的行。我们用内置的 node:test test runner 来测试。模拟数据源是 Readable.from，其中的 chunk 故意在行和 UTF-8 字符的中间切割：

// split-lines.test.mjs
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { Readable } from 'node:stream';
import { splitLines } from './split-lines.mjs';

async function collect(stream) {
  const result = [];
  for await (const item of stream) result.push(item);
  return result;
}

test('拼接两个 chunk 之间被切割的行', async () => {
  const source = Readable.from([
    Buffer.from('1,an,10,PAID\n2,bi'),
    Buffer.from('nh,20,FAILED\n3,chi,5,PAID'), // 最后一行没有 \n
  ]);

  const lines = await collect(source.pipe(splitLines()));

  assert.deepEqual(lines, ['1,an,10,PAID', '2,binh,20,FAILED', '3,chi,5,PAID']);
});

test('不破坏多字节 UTF-8 字符', async () => {
  const bytes = Buffer.from('1,Nguyễn Văn Hiệp,10,PAID\n');
  const source = Readable.from([bytes.subarray(0, 9), bytes.subarray(9)]); // 在字符"ễ"中间切割

  const lines = await collect(source.pipe(splitLines()));

  assert.deepEqual(lines, ['1,Nguyễn Văn Hiệp,10,PAID']);
});

运行测试：

node --test

✔ 拼接两个 chunk 之间被切割的行 (3.1ms)
✔ 不破坏多字节 UTF-8 字符 (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

在这个测试中，使用 .pipe() 仍然没问题，因为 Readable.from 在内存中运行，没有需要清理的 I/O。如果你删除 StringDecoder 并使用 chunk.toString()，第二个测试会失败，字符串会包含 �。

8. 常见错误和何时不应使用 stream

使用 on('data') 配合 async handler。 例如 stream.on('data', async chunk => await db.insert(chunk))。Event emitter 不会等待 Promise，所以 stream 仍然以最大速度推送数据，数千个 insert 语句并行运行。请使用 for await 或在 pipeline 中使用 async generator 步骤。

忘记处理错误。 没有监听器的 'error' event 会导致整个进程崩溃。对于 Promise 形式的 pipeline，你只需要在一个地方使用 try/catch。

假设 chunk 是一行或一条记录。 Chunk 按字节切割。请始终用 readline 或具有 tail buffer 的 Transform 来分行，如第 6.2 节所示。

设置 highWaterMark 过大。 设置 100 MB 来“更快地读取”几乎没有帮助，反而会失去内存方面的好处。64 KiB 到 1 MiB 的值通常是足够的。

将结果收集到一个大数组中。 形如 for await (...) rows.push(row) 然后才处理的代码正好回到了 readFile 的问题。请在接收数据时立即处理或写出。

注意： Stream 并不总是最佳选择。对于几 KB 的配置文件，或者当你需要一次性处理所有数据时（排序全部、解析单个 JSON 文件），readFile 更简单快速。只有当数据大小很大、事先不知道大小或来自网络时才使用 stream。

9. 结论

你已经看到 readFile 的 RAM 消耗是文件大小的多倍。Stream 通过逐块处理保持内存稳定。要点：

• 使用 createReadStream 配合 for await 来按块读取，以及 readline 来逐行读取。
• 尊重 backpressure：检查 write() 的返回值，或让 pipeline 处理这个问题。
• 使用来自 node:stream/promises 的 pipeline 代替 .pipe()，这样错误会通过各个 stream 传递，资源会被正确清理。
• 编写 Transform 来处理不完整的行和不完整的 UTF-8 字节，然后用故意切割的 chunk 进行测试。

有了这些内置工具，你可以过滤、转换和压缩多 GB 的文件，同时进程仅使用几十 MB 的 RAM。