Node.js 深入：流背压、取消与资源诊断

运行时与并发边界

前置为模块、Promise、Node文件及测试。使用Node24.21.0，项目只有标准库，不要求框架。JavaScript回调通常在事件循环线程执行；文件I/O可由运行时与系统异步完成，但JSON.parse、排序等CPU工作仍会阻塞当前线程。worker_threads用于CPU并行隔离，不能通过把同步循环包成async让它自动并行。进程拥有独立堆，Worker线程也有自己的JS执行环境，通信和数据复制/转移有成本。

正常退出依赖活动句柄和待执行工作；pending Promise本身不保证进程存活。不要用无意义的setInterval阻止退出，也不要在打印错误后立刻process.exit导致输出或清理被截断；本CLI设置exitCode并在finally移除SIGINT监听器。平台信号行为存在差异，核心取消协议用AbortController直接测试，不把Windows终端Ctrl+C行为写成Linux信号完全等价。

流、背压与编码

readFile一次装入整个文件；流允许按块消费并将速度反馈给上游。手动write返回false时应等待drain，不能继续无限写；pipeline连接源、异步生成器和目标，统一传播错误和关闭。本例sink的highWaterMark=1且异步调用done，下一批消费受背压约束。highWaterMark是缓冲阈值，不是全进程内存硬上限；源块、解码缓冲、解析对象和系统缓冲仍占内存。

文件块不是行，也不保证UTF-8字符完整。TextDecoder以stream:true保留跨块字节，结束时decode()检查未完成序列，fatal:true拒绝坏编码。先找换行，再解析JSON；CRLF的CR只在行末移除。每行最多4096个UTF-16单元（含可选CR，不含LF）、name1–80码点、value为0–100000安全整数，最多10000条；空文件成功，空行失败，末尾一个LF不额外产生空记录。输入格式是JSONL，不是任意连续JSON或通用日志协议。

完整流处理项目

三个实现文件加测试位于examples/node/stream-lab，PowerShell/Bash命令相同。输出只有汇总成功才写入stdout，不留下中间部分统计；缺/多参数退出2，文件/编码/输入/取消错误退出1并写stderr，成功退出0。

summary.mjs

import {Writable} from 'node:stream';
import {pipeline} from 'node:stream/promises';

export async function summarize(source,{signal,limit=10000}={}){
 if(!Number.isInteger(limit)||limit<1||limit>10000)throw new RangeError('limit must be 1..10000');
 const result={count:0,sum:0};
 async function* records(chunks){
  const decoder=new TextDecoder('utf-8',{fatal:true});let buffer='';
  function parse(line){
   if(line.length>4096)throw new Error('line length must be 1..4096');
   if(line.endsWith('\r'))line=line.slice(0,-1);
   if(!line||line.length>4096)throw new Error('line length must be 1..4096');
   const value=JSON.parse(line);
   if(!value||Array.isArray(value)||typeof value.name!=='string'||[...value.name].length<1||[...value.name].length>80||!Number.isSafeInteger(value.value)||value.value<0||value.value>100000)throw new Error('invalid record');
   return value;
  }
  for await(const chunk of chunks){
   buffer+=decoder.decode(chunk,{stream:true});
   let newline;
   while((newline=buffer.indexOf('\n'))>=0){const line=buffer.slice(0,newline);buffer=buffer.slice(newline+1);yield parse(line);}
   if(buffer.length>4096)throw new Error('line length must be 1..4096');
  }
  buffer+=decoder.decode();
  if(buffer)yield parse(buffer);
 }
 const sink=new Writable({objectMode:true,highWaterMark:1,write(record,_encoding,done){
  if(result.count>=limit){done(new Error('record limit exceeded'));return;}
  result.count++;result.sum+=record.value;setImmediate(done);
 }});
 await pipeline(source,records,sink,{signal});
 return result;
}

cli.mjs

import {createReadStream} from 'node:fs';
import {summarize} from './summary.mjs';
const args=process.argv.slice(2);
if(args.length!==1){console.error('usage: node cli.mjs input.jsonl');process.exitCode=2;}
else{
 const abort=new AbortController();const interrupt=()=>abort.abort();process.once('SIGINT',interrupt);
 try{console.log(JSON.stringify(await summarize(createReadStream(args[0]),{signal:abort.signal})));}
 catch(error){console.error('error: '+error.message);process.exitCode=1;}
 finally{process.removeListener('SIGINT',interrupt);}
}

sample.jsonl

{"name":"reading","value":20}
{"name":"testing","value":30}

summary.test.mjs

import {test} from 'node:test';
import assert from 'node:assert/strict';
import {Readable} from 'node:stream';
import {summarize} from './summary.mjs';
test('UTF-8跨块、CRLF、最后无换行与精确总数',async()=>{
 const bytes=Buffer.from('{"name":"中","value":2}\r\n{"name":"x","value":3}');
 const chunks=[...bytes].map(b=>Buffer.from([b]));
 assert.deepEqual(await summarize(Readable.from(chunks)),{count:2,sum:5});
 assert.deepEqual(await summarize(Readable.from([])),{count:0,sum:0});
});
test('格式、坏编码、空行、边界和源错误不输出部分成功',async()=>{
 for(const data of ['\n','{}','{"name":"x","value":1.5}','{"name":"x","value":-1}','{"name":"x","value":100001}', 'x'.repeat(4097)])await assert.rejects(summarize(Readable.from([Buffer.from(data)])));
 await assert.rejects(summarize(Readable.from([Buffer.from([255])])));
 await assert.rejects(summarize(Readable.from([Buffer.from('{"name":"x","value":1}\n'.repeat(3))]),{limit:2}),/record limit/);
 const broken=Readable.from((async function*(){yield Buffer.from('{"name":"x","value":1}\n');throw new Error('source failed');})());
 await assert.rejects(summarize(broken),/source failed/);
});
test('预取消和运行中取消均拒绝且关闭源',async()=>{
 const controller=new AbortController();controller.abort();const source=Readable.from([Buffer.from('{}')]);
 await assert.rejects(summarize(source,{signal:controller.signal}),{name:'AbortError'});assert.equal(source.destroyed,true);
 const live=new AbortController();const stream=new Readable({read(){this.push(Buffer.from('{"name":"x","value":1}\n'));live.abort();}});
 await assert.rejects(summarize(stream,{signal:live.signal}),{name:'AbortError'});assert.equal(stream.destroyed,true);
});

从仓库根执行node examples/node/stream-lab/cli.mjs examples/node/stream-lab/sample.jsonl，预期一行{"count":2,"sum":50}。执行node --test examples/node/stream-lab/summary.test.mjs，预期3组通过：逐字节分割中文/CRLF/无末尾LF、格式与源错误、预取消与运行中取消且源destroyed。读不存在文件应只有error并非零退出；测试代码断言的是抛错，不以console.assert当硬失败。

性能、GC与诊断

先用相同输入和相同结果检查正确性，再测量处理时间、吞吐及内存。Node --cpu-prof cli.mjs sample.jsonl生成CPU采样文件，可在DevTools载入；--heap-prof采集分配采样，--inspect启动调试接口，实际使用时只绑定127.0.0.1。采样文件可能包含路径或业务信息，不应随意公开。可用process.memoryUsage区分heapUsed、external、arrayBuffers和rss；GC释放JS对象不保证操作系统RSS立即下降。

本例小文件耗时可能主要是启动成本，不能得出流一定更快。比较readFile时须保持坏编码/行长/条数校验语义一致；快速但忽略边界的版本不是等价优化。生产中若单条JSON极大，即使流式读取也会在解析该条时占用CPU与内存，因此先定义输入上限，再讨论并发和吞吐。CPU分析结论只针对实测负载，不承诺固定倍数。

ERR_MODULE_NOT_FOUND先查相对路径与.mjs后缀；ENOENT先查启动目录；ERR_STREAM_PREMATURE_CLOSE先找谁提前关闭；AbortError是取消协议结果，不能当有效空文件成功。npm锁和版本诊断见前端工程化，HTTP服务的部署和授权另按实际应用设计，不把本CLI称为完整服务框架。

练习

1. 为什么逐字节中文能成功，但255字节失败？2. 删除done延迟后，如何仍验证背压而不靠固定毫秒？3. 增加平均值时空文件如何定义？4. 若输出改为逐条写文件，应如何避免失败留下伪完整结果？

参考解法

解码器保留未完成UTF-8序列，255不是合法起始字节；结束时还需flush。用受控sink保存done回调，断言上游未越过约定队列，然后手动放行；时间不是同步协议。平均值要明确空集输出null或单独字段，不能除以0后把NaN序列化成误导结果。写独立临时文件、成功后发布并定义替换语义，同时处理失败清理与跨文件系统问题；不能仅把stdout重定向原输入文件，Shell可能先截断输入。

依据：Node Stream、Node24测试、中文Promise组织。官方当前主页可能显示更新主版本，本项目仅使用已在Node24实际验证的接口，版本专属接口须另查兼容性；核对2026-10-17。