Node.js 流处理进阶：内存管理、背压机制与真实故障处理
了解 Node.js 流如何与 Web Streams 相互配合，通过实际基准测试衡量内存节省效果，以及仅在高负载情况下才会出现的生产环境错误。
来源: Node.js Streams in Production: The Parts Most Tutorials Skip
许多关于 Node.js 流的入门教程都很好地讲解了基础内容：可读流、可写流、双向流、转换操作、pipe() 方法、背压机制，还有gzip 压缩演示，有时还会以视频范围请求服务器作为结尾示例。如果你已经看过其中一两个教程，就应该明白什么是流，以及为什么在演示中要把2GB的文件全部加载到内存中会带来诸多问题。
那些介绍很少提及的是接下来的内容。切换到流式处理后究竟能节省多少内存？请给出具体数字，而非含糊的“更少”这样的表述。Node 的原生流与 fetch 及浏览器所使用的 Web Streams API 是如何交互的？又该如何在自己的硬件上运行模型并直接输出流式数据，而无需通过托管 API 按令牌数量付费？还有，哪些两三个错误会在处理真实流量时，让在测试中表现良好的数据管道出现故障？
本文正好探讨了这些问题。它假设你已经了解什么是可读流。如果还不清楚，下面的简短概述能帮你快速掌握要点。后续的所有内容都将基于这些基础知识展开。
60秒速览
Readable -> produces data (fs.createReadStream, HTTP request body)
Writable -> consumes data (fs.createWriteStream, HTTP response)
Duplex -> both, independently (a TCP socket)
Transform -> consumes, changes, produces (zlib.createGzip)
Readable stream -> Transform stream -> Writable stream
数据在系统中是以小块形式传输的，而非作为一个整体。pipe()用于将多个数据流连接起来，并处理目标端无法跟上源端速度的情况——即背压问题。pipeline()也能实现同样的连接功能，但它还会将链中任意数据流出现的错误传递出去，并在出现故障时确保进行清理工作，正因如此，在任何需要长期运行的服务中都应该使用它，而非仅用于一次性快速脚本。
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
await pipeline(
createReadStream('./input.txt'),
createGzip(),
createWriteStream('./input.txt.gz')
);
如果那个例子你已经完全理解了，可以直接跳过。接下来介绍的内容是在首次将基于流处理的代码投入生产环境之前需要了解的知识。
这些文章中从未提及的数字
大多数关于流处理的文章都声称缓冲方式“会占用大量内存”，而流式处理则“占用内存少得多”，但却从未给出具体的数值作为依据。为了解决这个问题，我们生成了一个501MB的纯文本文件，并在同一台机器上依次对两种处理方式各进行了两次测试。
// buffered.js: read the whole file, then scan it
import { readFileSync } from 'node:fs';
const before = process.memoryUsage().rss;
const data = readFileSync('./big.txt', 'utf8');
const afterRead = process.memoryUsage().rss;
let count = 0;
for (let i = 0; i < data.length; i++) if (data[i] === 'q') count++;
console.log({
rssBeforeMB: (before / 1024 / 1024).toFixed(1),
rssAfterReadMB: (afterRead / 1024 / 1024).toFixed(1),
count
});
// streamed.js: read in 64KB chunks, scan as it arrives
import { createReadStream } from 'node:fs';
let peakRss = 0;
const timer = setInterval(() => {
peakRss = Math.max(peakRss, process.memoryUsage().rss);
}, 10);
let count = 0;
const stream = createReadStream('./big.txt', { encoding: 'utf8', highWaterMark: 64 * 1024 });
stream.on('data', (chunk) => {
for (let i = 0; i < chunk.length; i++) if (chunk[i] === 'q') count++;
});
stream.on('end', () => {
clearInterval(timer);
console.log({ peakRssMB: (peakRss / 1024 / 1024).toFixed(1), count });
});
以下是在使用Node.js 22对500MB的ASCII文件进行测试时，两种处理方式各运行两次的实际结果：
Approach Peak RSS Wall time Notes
--------------------------------------------------------------------
readFileSync (buffered) ~539 MB 1.24 - 1.76s Full file materialized as a string
createReadStream, 64KB chunks ~52 MB 1.56 - 1.67s Roughly 10x less peak memory
createReadStream, 1MB chunks ~50 MB 1.25 - 1.34s Bigger chunks, same ballpark memory
有几点需要注意的地方，若不说明可能会让结果产生误导：此测试仅针对单台机器、单个文件以及单一工作负载——即一次简单的字符扫描。两种方法的实际运行时间大致相当，流式处理方式在某些情况下甚至略快，这有点出乎意料。最初的假设是处理大量小数据块的额外开销会带来更明显的性能下降。而内存使用情况则与预期高度吻合：缓冲式处理的内存峰值接近文件总大小，而流式处理的内存峰值则较低，且几乎不受文件大小影响而保持稳定。这种对比恰恰体现了使用流式处理的优势，所有信息都汇总在一张表格中——通过用上述两个脚本对文件进行测试，也可以轻松独立验证这一点。
你自己的。实际意义在于：对于由性能足够的机器处理的单个请求，只需将整个文件读入内存通常就足够了。真正的问题从来不是单独的某个请求——真正关键的是随之而来的问题：当有100个这样的请求同时运行时会发生什么？在流式处理方式下，随着并发数的增加，内存使用量几乎不会变化，因为每个请求只保留一段大小的缓冲区，而非一次性处理全部数据。
Node streams与Web Streams现在属于同一系列
多年来，Node.js 流与 WHATWG Streams API（ReadableStream、WritableStream、TransformStream——这些类是浏览器和 Node 环境中 fetch() 函数的实现基础）看似属于两个互不重叠的独立生态系统。但这一看法已不再准确。node:stream/web 模块可直接使用 WHATWG 的这些类，而 Node 自带的 Readable、Writable 和 Duplex 类也都具备了 .toWeb() 和 .fromWeb() 方法，根据当前版本的 Node.js，它们的稳定性等级为 2（“稳定”）。
一旦在 Node 服务中调用 fetch()，这一点就显得尤为重要，因为返回的 response.body 是 Web 环境下的 ReadableStream，而非 Node 环境下的 Readable。
import { createWriteStream } from 'node:fs';
import { Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
const response = await fetch('https://example.com/large-report.csv');
if (!response.ok || !response.body) {
throw new Error(`Download failed with status ${response.status}`);
}
// response.body is a Web ReadableStream, so convert the Node side instead
const fileStream = Writable.toWeb(createWriteStream('./report.csv'));
await response.body.pipeTo(fileStream);
console.log('Download saved without buffering the response in memory');
反向转换同样顺畅——你可以将 Node 的 Readable 对象转换为任何基于 Web-stream 的 API 都能接受的流：
import { createReadStream } from 'node:fs';
import { Readable } from 'node:stream';
const webStream = Readable.toWeb(createReadStream('./large-file.bin'));
// Now this works with fetch's request body, a browser upload API,
// or any library built against the WHATWG interface instead of Node's.
await fetch('https://uploads.example.com/put', {
method: 'PUT',
body: webStream,
duplex: 'half' // required by fetch when the body is a stream
});
初学者常会遇到的一个问题是：如果在不设置 duplex: 'half' 的情况下将流式请求体传递给 fetch，它会抛出异常。之所以如此，是因为并非所有地方都支持流式请求体，因此规范要求必须明确启用该功能。这虽是个小问题，但一旦遇到可能需要花费15分钟时间解决，之后就再不会出现类似问题。
真正的优势在于，现在你的 Node 后端、边缘函数以及浏览器代码大多可以使用相同的流处理逻辑，而无需再维护一个针对 fetch 的版本以及另一个基于 Node 的 fs 和 http 模块构建的 pipe() 版本。
stream.compose()：一个返回单个流的管道
从技术上讲，stream.compose()早在 Node.js 16.9.0 版本就已存在，但在很长一段时间内都处于实验状态，直到最近某个 Node.js 发布版本才最终稳定下来。既然其 API 已经确定，就值得仔细研究一下。
其设计理念是：无需编写带有 _transform 和 _flush 方法的完整 Transform 子类，只需编写一个异步生成器函数，compose() 就会将这一系列操作合并为一个 Duplex 流，你可以像处理其他流一样对其使用管道操作。
import { compose } from 'node:stream';
import { createReadStream, createWriteStream } from 'node:fs';
async function* toUpperCase(source) {
for await (const chunk of source) {
yield chunk.toString().toUpperCase();
}
}
async function* onlyNonEmptyLines(source) {
let leftover = '';
for await (const chunk of source) {
const lines = (leftover + chunk).split('\n');
leftover = lines.pop() ?? '';
for (const line of lines) {
if (line.trim()) yield line + '\n';
}
}
if (leftover.trim()) yield leftover;
}
const pipeline = compose(
createReadStream('./raw.txt', { encoding: 'utf8' }),
onlyNonEmptyLines,
toUpperCase
);
pipeline.pipe(createWriteStream('./clean.txt'));
在编写两个独立的Transform类的同时设置该功能——包括为处理行分割部分而实现的手动缓冲逻辑——即可获得相同的底层行为，但代码结构明显更简洁，且代码的读取顺序与数据实际流动的顺序一致。compose()在端点处理上也十分灵活：若将普通的AsyncIterable作为第一个参数，它就会变成可读取的Duplex；若将AsyncFunction作为最后一个参数，它则会变成可写入的接收端。这意味着，如果偏好这种风格，完全可以仅使用异步生成器来构建整个处理流程。
以上这些都不是删除那些已经能正常工作的 Transform 类的理由。但对于新代码——尤其是那些需要“解析此数据、过滤那部分内容、重新整理结构”的任务——使用异步生成器结合 compose() 方法，通常可以避免编写多个功能几乎完全相同的 Transform 子类。
从本地模型中流式输出标记
教程中的几乎所有“现实世界数据流”示例都只涉及三种情况：文件下载、gzip压缩以及视频片段请求。这些确实是合理的应用场景，但还有一种更新的模式同样适用于流式处理模型，却很少被提及：从大型语言模型中逐个获取标记形式的输出。
由于真正可复制粘贴的示例不应依赖任何人的 API 密钥，下方的示例是与本地运行的 Ollama 进行交互，而非使用托管服务：
# one-time setup
ollama pull llama3.2
ollama serve # usually already running as a background service
默认情况下，Ollama 的 /api/generate 接口会以换行分隔的 JSON 格式（application/x-ndjson）输出响应：每行一个 JSON 对象，每个对象包含 response 字段和 done 标志，最后一行会显示 "done": true。去掉格式化后，这其实只是用于简单文本解析的可读流——本文其他部分所使用的基于分块的推理方式可直接沿用，无需修改。
// server.js: proxies Ollama's token stream to the browser over SSE
import { createServer } from 'node:http';
const server = createServer(async (request, response) => {
if (request.url !== '/ask') {
response.statusCode = 404;
response.end('Not found');
return;
}
response.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive'
});
const ollamaResponse = await fetch('http://localhost:11434/api/generate', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
model: 'llama3.2',
prompt: 'Explain backpressure in one short paragraph.',
stream: true
})
});
if (!ollamaResponse.body) {
response.end();
return;
}
// Web ReadableStream from fetch -> async iterate it directly (Node 18+)
let buffered = '';
for await (const chunk of ollamaResponse.body) {
buffered += Buffer.from(chunk).toString('utf8');
const lines = buffered.split('\n');
buffered = lines.pop() ?? '';
for (const line of lines) {
if (!line.trim()) continue;
const event = JSON.parse(line);
response.write(`data: ${JSON.stringify({ text: event.response })}\n\n`);
if (event.done) {
response.write('event: done\ndata: {}\n\n');
response.end();
return;
}
}
}
// client disconnected mid-stream? stop pulling tokens we can't deliver
request.on('close', () => {
if (!response.writableEnded) response.end();
});
});
server.listen(3000, () => console.log('Ask at http://localhost:3000/ask'));
如果您想查看 Ollama 输出的原始 NDJSON 行，且不带 SSE 框架封装，其格式如下：
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.121Z","response":"Back","done":false}
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.132Z","response":"pressure","done":false}
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.166Z","response":" happens...","done":true,"done_reason":"stop"}
初次接触时，有两个细节容易让人犯错。首先，在当前的 Node.js 版本中，fetch 的 response.body 本身就已经是异步可迭代对象，因此无需专门为这种情况调用 Readable.fromWeb()——不过如果某些下游库期望的是传统的 Node Readable 对象，该辅助函数仍然可用。其次，NDJSON 的数据行可能跨越多个数据块边界，这与流式解析 CSV 时出现的风险完全相同：单个 fetch 数据块并不一定与单行 JSON 数据对应。正因如此，上面的示例才会对不完整的行进行缓冲处理，而非直接对接收到的数据调用 JSON.parse(chunk)。
如果你更愿意使用托管的 API——比如 OpenAI、Anthropic 或类似服务——整体流程几乎不会改变：你仍然需要接收分块的 HTTP 响应，然后以 SSE 的形式将其转发给客户端。本地版的 Ollama 优势在于无需账户或密钥即可立即运行，因此在开发阶段构建此类功能的客户端时是个不错的选择；等到需要生产级输出质量时，再将相同代码用于托管模型即可。
直接将文件流式上传到对象存储
将文件“发送到S3”时，几乎从来不会出现文件小到可以轻松放入内存的情况。@aws-sdk/lib-storage包的存在正是为了解决普通PutObjectCommand在发起请求前就需要知道内容长度的问题。它的Upload辅助函数接受数据流作为输入，自行判断是使用单次请求还是分块上传更为合适，并在传输过程中提供进度反馈。
import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
import { createReadStream } from 'node:fs';
const s3 = new S3Client({ region: 'us-east-1' });
async function uploadFile(localPath, bucket, key) {
const upload = new Upload({
client: s3,
params: {
Bucket: bucket,
Key: key,
Body: createReadStream(localPath)
}
});
upload.on('httpUploadProgress', (progress) => {
console.log(`${key}: ${progress.loaded} / ${progress.total ?? '?'} bytes`);
});
await upload.done();
console.log(`${key} uploaded`);
}
如果不想拥有真实的AWS账户，也可以通过MinIO来实验，它提供了一个与S3兼容的对象存储服务，只需一条命令即可在本地启动。上述的@aws-sdk/lib-storage代码无需任何修改即可直接用于MinIO，只需将客户端指向MinIO的端点而非AWS即可：
docker run -p 9000:9000 -p 9001:9001 \
-e "MINIO_ROOT_USER=admin" \
-e "MINIO_ROOT_PASSWORD=admin12345" \
minio/minio server /data --console-address ":9001"
const s3 = new S3Client({
region: 'us-east-1',
endpoint: 'http://localhost:9000',
forcePathStyle: true, // required for MinIO's path-style URLs
credentials: { accessKeyId: 'admin', secretAccessKey: 'admin12345' }
});
对于容量超过几十兆字节的文件，建议采用这种方式：在将文件传递给 SDK 之前，避免将其全部加载到 Buffer 中。应使用 createReadStream() 提供数据流，让 Upload 处理后续操作——无论文件是 10MB 还是 10GB，内存使用量都不会发生变化。
不缓冲请求体即可解析上传内容
文件上传的另一方是必须接收这些文件的服务器，而意外缓冲问题最常出现在这里——通常是因为某些用于解析请求体的中间件并未考虑流式处理，在处理程序有机会操作数据之前就悄悄将整个多部分请求读入了内存。
busboy以渐进方式处理多部分表单解析：一旦某个文件字段的字节开始传入，它会立即为该字段返回一个可读取的数据流，而无需等待整个请求全部到达：
import { createServer } from 'node:http';
import { createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import busboy from 'busboy';
const server = createServer((request, response) => {
if (request.method !== 'POST' || request.url !== '/upload') {
response.statusCode = 404;
response.end();
return;
}
const bb = busboy({ headers: request.headers, limits: { fileSize: 200 * 1024 * 1024 } });
const uploads = [];
bb.on('file', (fieldName, fileStream, info) => {
const destinationPath = `./uploads/${Date.now()}-${info.filename}`;
uploads.push(
pipeline(fileStream, createWriteStream(destinationPath)).catch((error) => {
fileStream.resume(); // drain on failure so the request can still finish
throw error;
})
);
});
bb.on('close', async () => {
try {
await Promise.all(uploads);
response.writeHead(200, { 'Content-Type': 'application/json' });
response.end(JSON.stringify({ status: 'ok' }));
} catch (error) {
response.statusCode = 500;
response.end(JSON.stringify({ status: 'error', message: error.message }));
}
});
request.pipe(bb);
});
server.listen(3000);
该处理程序从不会一次性存储完整的请求体。字节从套接字传来，经过busboy的解析器后直接写入磁盘。正是这种设计差异，使得某些服务能够轻松处理多个大文件同时上传的情况，而另一些服务则在有两个用户同时上传视频时就会崩溃。
测试自定义转换流
手动创建的 Transform 流极易出现问题：没人愿意为它们编写测试，大家都认为“反正能正常工作”，结果某个 _flush 的边缘情况导致功能出错，而直到生产环境才发现。不过，为这类流编写测试其实并没有想象中那么麻烦。你无需创建真实的文件或 HTTP 服务器，只需结合使用 Readable.from() 和 stream/promises 即可完成测试。
// uppercase-transform.js
import { Transform } from 'node:stream';
export function createUppercaseTransform() {
return new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
}
});
}
// uppercase-transform.test.js
import { describe, it, expect } from 'vitest'; // works the same with Jest
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createUppercaseTransform } from './uppercase-transform.js';
async function collect(readable) {
const parts = [];
for await (const chunk of readable) parts.push(chunk.toString());
return parts.join('');
}
describe('uppercase transform', () => {
it('uppercases each chunk', async () => {
const source = Readable.from(['hello ', 'world']);
const transform = createUppercaseTransform();
source.pipe(transform);
expect(await collect(transform)).toBe('HELLO WORLD');
});
it('propagates errors from upstream instead of hanging', async () => {
const source = new Readable({
read() {
this.destroy(new Error('upstream failed'));
}
});
const transform = createUppercaseTransform();
await expect(pipeline(source, transform, async function drain(it) {
for await (const _ of it) { /* discard */ }
})).rejects.toThrow('upstream failed');
});
});
该文件中的第二个测试比乍看之下更重要。它能够确认上游抛出的错误确实会传递出去，而不会让转换流在处理过程中因承诺状态永远无法解决而卡住。这种无声的停滞正是那些看似正常工作的处理流程在投入实际使用后变成无法处理的请求的最常见原因。
轻量级指标采集器
只需检查流水线的运行状况，并不需要完整的可观测性平台。通常，一个能够统计流经数据的字节数和分块数的小型PassThrough流就足够了，而且它可以直接插入到现有的流水线中而不会影响其他部分：
import { PassThrough } from 'node:stream';
export function createMetricsTap(label) {
let bytes = 0;
let chunks = 0;
const startedAt = process.hrtime.bigint();
const tap = new PassThrough();
tap.on('data', (chunk) => {
bytes += chunk.length;
chunks += 1;
});
tap.on('end', () => {
const seconds = Number(process.hrtime.bigint() - startedAt) / 1e9;
console.log(`[${label}] ${chunks} chunks, ${(bytes / 1024 / 1024).toFixed(2)} MB, ` +
`${(bytes / 1024 / 1024 / seconds).toFixed(2)} MB/s`);
});
return tap;
}
await pipeline(
createReadStream('./large-file.bin'),
createMetricsTap('upload'),
createGzip(),
createWriteStream('./large-file.bin.gz')
);
如果想要比console.log更强大的输出功能，可以用prom-client的Counter或Histogram来替换'data'和'end'处理函数中的逻辑。采集器的结构本身保持不变：它只是流水线中的一个普通流，负责报告流经的数据情况。
仅在负载过高时才会显现的三个错误
未被处理的 'error' 事件会导致整个进程崩溃。 如果某个数据流在未设置 'error' 监听器的情况下抛出错误，它不仅会使自身的请求失败。在 Node.js 中，EventEmitter 上的 'error' 事件若无人处理就会直接抛出异常，而如果没有代码捕获该异常，进程将会终止。仅仅一次格式错误的上传就足以让整个服务下线，因为手动连接的 pipe() 链中三个数据流中有两个设置了错误监听器。pipeline() 的存在正是为了解决这类缺陷，因此建议优先使用它，或者确保自己手动创建的每个数据流都设置独立的 .on('error', ...) 监听器，绝不能有例外。
重复使用流实例会导致监听器泄漏。假设你在每次请求时都会通过一个函数来创建流，但该函数之外的某些代码却保留了对这些流的引用，并“以防万一”地附加了自己的监听器。最终你会在日志中看到 MaxListenersExceededWarning 警告，这是 Node 用来提示内存泄漏正在发生的信号。每个请求都需要使用新创建的流，一旦请求处理完毕，其生命周期之外的任何代码都不应再持有这些流的引用。
当客户端离开时，不会有人去销毁数据流。这正是会在实际生产环境中造成损失的问题。如果用户在下载过程中关闭了浏览器标签页，而你没有监听 response.on('close', ...) 事件来调用源数据流的 .destroy() 方法，服务器仍会继续从磁盘读取数据；更糟糕的是，它可能会保持数据库游标处于打开状态或维持上游HTTP连接，而这一切都是为了那个已经离开的客户端。在足够的并发负载下，服务器就会这样悄无声息地耗尽文件描述符或连接池，而日志中却看不出明显原因。解决方法始终是同样的三行代码：监听目标端的关闭信号，随即销毁源数据流，并将这种处理方式作为每个基于数据流的路由的默认操作，而非仅在出现问题后才去想起。
何时使用哪种 API
pipe()
Simple, two-or-three-stream chains. Handles backpressure.
Error handling is your responsibility per-stream.
pipeline()
Anything running in production. Handles backpressure,
forwards errors from every stream, cleans up on failure.
Has a promise-based version and supports AbortSignal.
stream.compose()
New pipelines built from async generators instead of
Transform subclasses. Returns one Duplex you can pipe
like any other stream. Stable as of recent Node.js releases.
node:stream/web (ReadableStream/WritableStream/TransformStream)
Anywhere you're mixing Node streams with fetch(), browser
code, or a library built against the WHATWG standard.
Convert with .toWeb() / .fromWeb() at the boundary.
这些方法之间并无相互替代的关系。它们各自解决系统中数据流不同环节的问题。如今，常见的做法是让单个处理流程从 fetch 返回的 Web ReadableStream 开始，对数据进行转换，再通过用 compose() 组合的变换函数处理，最后将结果传递给 pipeline() 进行写入操作。过去以这种方式混合使用 API 会让人觉得不寻常，但如今这已是一种相当常见的模式。
总结
即便如此，仍需先理解背压机制、highWaterMark以及四种基础流类。在运用上述任何技术之前，这些基础知识都是必不可少的。掌握了这些背景知识后，真正的工程工作才刚刚开始：需要用具体数据而非模糊说法来验证内存使用情况的改善；要记住基于fetch的流与Node流只需一次.toWeb()转换即可实现互操作；此外，在编写路由时还需预先考虑客户端在请求过程中断开连接的情况，将其视为标准部分而非事后才需要修补的细节。
如果在不同的硬件上运行前面提到的基准测试示例，且结果与此处显示的有明显差异，那其实是一个值得分享的有用数据，而非可以忽略的异常值——因为当在多台机器上对比后，吞吐量和内存使用量的数值才会更加可靠。
参考资料
- 介绍经典 Stream API 的 Node.js 文档：https://nodejs.org/api/stream.html
- 介绍 Web Streams API 的 Node.js 文档：https://nodejs.org/api/webstreams.html
- Ollama 的流式接口参考文档：https://docs.ollama.com/api/streaming
- npm 上的
@aws-sdk/lib-storage包信息：https://www.npmjs.com/package/@aws-sdk/lib-storage busboy项目仓库：https://github.com/mscdex/busboy