异步迭代与 Streams：一段一段消费数据
同步迭代器要求 next 立刻给出结果。若下一段数据要等网络或文件读取，就需要异步迭代器：next 返回 Promise，完成后得到 { value, done }。消费者通过 for await...of 等待每一段。
先用异步生成器观察拉取节奏
async function* blocks() {
try {
for (let n = 1; n <= 3; n++) {
console.log('准备', n);
yield await Promise.resolve(`块${n}`);
}
} finally { console.log('关闭'); }
}
(async () => {
for await (const block of blocks()) {
console.log('消费', block);
break;
}
})();
// 准备 1 → 消费 块1 → 关闭没有准备第二块，因为消费者没有再次请求。这个节奏提供最基本的按需拉取：循环体里 await 慢处理，下一次 next 就会晚发生。但如果生成器内部已经启动大量后台工作，外面按需迭代也无法撤销提前启动的那些任务。
Symbol.asyncIterator 是异步协议入口。for await 也可以消费同步可迭代对象，并等待其值中的 Promise；这是协议适配，不意味着普通数组变成多线程。同步来源的拒绝值还有一个与运行时版本有关的关闭细节，下面单独比较。
等待中的 next 不会被 return 立即打断
let finish;
const waiting = new Promise(resolve => { finish = resolve; });
async function* oneBlock() {
try { yield await waiting; }
finally { console.log('执行清理'); }
}
const iterator = oneBlock();
const next = iterator.next();
const closing = iterator.return();
console.log('已经请求关闭');
finish('数据到达');
console.log((await next).value); // 数据到达
await closing; // 清理随后完成，日志与恢复微任务的细节不作跨环境排序承诺异步生成器把对 next、return 等方法的请求排队处理。当前 next 还停在 await waiting，后来的 return 不会神奇地让这个 Promise 取消；先完成当前请求，才继续处理关闭请求。若 waiting 永不结束，关闭也可能一直等。把 AbortSignal 传给底层等待操作，才能让它通过拒绝或正常停止尽快回到生成器自己的 finally。
这也解释了为什么循环体内检查 signal.aborted 不是完整取消方案：如果循环正在等待下一块，循环体根本还没有运行机会。
同步来源的拒绝值：规范与旧运行时要分开看
在 Node 22.21 这样的运行时中，同步生成器 yield 一个拒绝 Promise，再由 for await 适配消费，拒绝可能离开适配层而不执行生成器内部 finally。较新的规范已经要求这个拒绝路径尝试关闭同步迭代器，不能把旧表现背成语言永恒规则。
以下片段特意输出实际环境的观测值；Node 22.21 输出 false，实现新关闭规则的环境应输出 true：
let closed = false;
function* syncSource() {
try { yield Promise.reject(new Error('读取失败')); }
finally { closed = true; }
}
try { for await (const value of syncSource()) {} }
catch (error) { console.log(error.message, closed); }
// Node 22.21：读取失败 false；采用新关闭规则的环境：读取失败 true需要覆盖旧环境时，可以让 async generator 在自身 try 中 await 操作，失败便明确发生在自身控制流里：
let closed = false;
async function* asyncSource() {
try { yield await Promise.reject(new Error('读取失败')); }
finally { closed = true; }
}
try { for await (const value of asyncSource()) {} }
catch (error) { console.log(error.message, closed); } // 读取失败 true两段看上去都在“产生 Promise”，但错误经过的位置不同。判断资源能否释放，要跟踪实际执行路径，并把运行时版本纳入验证。
Streams 增加队列与上下游管理
Web Streams 主要有 ReadableStream、WritableStream、TransformStream。它们不是 ECMAScript 的核心对象，而是宿主提供的数据流 API。生成器解释怎样按需获取值，Streams 还明确排队、锁定、取消、错误与背压传播。
(async () => {
let n = 1;
const stream = new ReadableStream({
pull(controller) {
controller.enqueue(n++);
if (n === 4) controller.close();
},
cancel(reason) { console.log('源被取消', reason); }
}, { highWaterMark: 0 });
const reader = stream.getReader();
try {
console.log((await reader.read()).value); // 1
await reader.cancel('已经足够'); // 源被取消 已经足够
} finally { reader.releaseLock(); }
console.log(stream.locked); // false
})();pull 在需要更多数据时调用；这里 highWaterMark 为 0，使示例不预先排队。普通流可以预取，所以不要把每次 pull 与每次 read 机械地一一对应。controller.desiredSize 表示队列还有多少期望容量；生产源需要配合这个信号，才不会无限 enqueue。高水位不是强制上限，更不是“超出就自动停止一切”的保险开关。
desiredSize 是反馈，生产者还得服从它
队列使用的量度由策略决定：默认可能按块数计，也可以按每块 byteLength 计。desiredSize = highWaterMark - 当前排队总大小；同样是 2 个块，每块 1 字节与每块 10 MB，对内存的意义不同。因此高水位值必须结合 size 函数解释。
可拉取来源能在没有需求时不调用底层读取；主动推送来源则需要暂停/恢复功能，例如收到背压后暂停设备推送。若来源完全不能暂停，就必须选择有界缓存、丢弃、采样或失败等策略。单纯包一层 ReadableStream 不会把不可控来源变得可控，持续无视 desiredSize 继续 enqueue 仍可能让内存增长。
下面用一个尚未完成的写入观察写侧压力，避免真实慢设备带来的随机性：
let finish;
const gate = new Promise(resolve => { finish = resolve; });
const stream = new WritableStream({ write(chunk) { return gate; } }, {
highWaterMark: 1,
size() { return 1; }
});
const writer = stream.getWriter();
console.log(writer.desiredSize); // 1
const written = writer.write('一块');
console.log(writer.desiredSize); // 0
finish();
await written;
await writer.ready;
console.log(writer.desiredSize); // 1
await writer.close();
writer.releaseLock();写入占用了期望容量，完成后容量恢复。ready 表示背压解除、可以继续推进，不表示整个文件已经写完或已经持久化；某次 write 的 Promise 反映该次底层写入完成，close 的 Promise 反映关闭过程完成。底层实现到底何时算“写入完成”，还取决于它自己的契约，不能把队列可用误当磁盘已同步。
锁、取消、关闭不是同一件事
getReader() 锁定流，同一时刻不能再交给另一个 reader 或管道独占读取。releaseLock() 只释放所有权，不表示取消底层工作。已锁定流需要通过 reader 取消；直接 stream.cancel() 会因锁定而失败。
生产者 close() 表示正常结束，队列里已有数据仍可消费；消费者 cancel(reason) 表示不再需要，排队数据会被丢弃并通知底层源；controller.error(error) 表示失败，后续读取拒绝。三种终止原因应在资源管理中分别处理。
直接 for await...of stream 在 break 时通常会取消流。如果只是暂时停止消费而希望后续继续，在支持的环境可以使用 stream.values({ preventCancel: true })；这仍要求明确由谁持有后续读取责任。
管道负责传递压力，也传递失败
readable.pipeThrough(transform).pipeTo(writable) 把分块转换和写入串起来。pipeTo 返回 Promise，必须 await 或 catch。默认情况下，源失败会中止目标，目标失败会取消源；preventAbort、preventCancel、preventClose 可以修改传播，但修改之后对应的清理责任就落在调用方。
处理文本流还有编码边界：一段 Uint8Array 不保证刚好落在字符结尾，逐块 new TextDecoder().decode(chunk) 可能把拆开的 UTF-8 字符解坏。使用同一个 decoder 的流模式，或 TextDecoderStream 保留未完成字节。这里的“块”是传输单位，不是字符、JSON 对象或行的天然边界。
WritableStream 的 writer 提供 ready 与写入 Promise，可以等待下游容量与完成。Node 传统 streams 是另一套接口：例如 write 返回 false 后等待 drain；不能把 Web Streams 的 writer 语法直接套进去，虽然现代 Node 也提供 Web Streams。
一次读取拿到的块不必等于一条记录
假设文本来源依次发来 '甲\n乙' 与 '\n丙\n'，第一块包含一条完整记录和下一条的开头，第二块才补齐。正确的行处理器要保存尚未成行的尾部，把它与下一块连接，再逐条取出分隔符之前的文本。二进制协议同理：长度字段本身也可能跨块，需要先攒够头部，再判断是否攒够负载。
编码状态负责从字节恢复字符，记录状态负责从字符恢复行或 JSON 消息。它们可以分层串成管道，但不能互相替代。TextDecoderStream 能修复跨块 UTF-8 字符，不会自动修复跨块 JSON；对每个传输块直接 JSON.parse 仍然会失败。
把跨块记录变成可以运行的转换器
先只处理已经解码的字符串，以换行符作为记录分隔。每次 transform 收到一块，把它接到 pending；只交付已经找到换行符的记录，余下尾部继续保留。正常关闭时 flush 再交付没有换行符的最后一行。输入 '甲\n乙' 时先交付甲、保留乙；下一块 '\n丙' 补齐乙，最后关闭时交付丙。
function lines() {
let pending = '';
return new TransformStream({
transform(chunk, controller) {
pending += chunk;
let end;
while ((end = pending.indexOf('\n')) !== -1) {
controller.enqueue(pending.slice(0, end));
pending = pending.slice(end + 1);
}
},
flush(controller) {
if (pending !== '') controller.enqueue(pending);
}
});
}
const input = new ReadableStream({
start(controller) {
controller.enqueue('甲\n乙');
controller.enqueue('\n丙');
controller.close();
}
});
const output = [];
await input.pipeThrough(lines()).pipeTo(new WritableStream({
write(line) { output.push(line); }
}));
console.log(JSON.stringify(output)); // ["甲","乙","丙"]这里源的 start 一次放入两个小块，目的是固定分块边界，不演示生产端背压。transform 对应任意传输块，flush 对应正常输入结束；错误结束不能当成完整文件来补交尾部。连续两个换行会产生空行，末尾一个换行不会额外产生空记录。这是本例选择的协议，调用方可以作不同约定。
背压也不能限制 pending：如果输入永远没有换行，转换器迟迟不交付任何记录，内部字符串仍会不断增长。真正接收外部文本时，应约定最大行长度并在超限时失败；队列容量与解析器内部状态是两处不同的内存来源。Windows 文本还可能使用 CRLF，本节练习会要求处理这个变化。
练习
为什么生成器示例在 break 后没有“准备 2”？
releaseLock()是否等于取消底层网络请求？async generator 正等待一项永不完成的 Promise 时，调用 return 是否会立即中断等待？为什么两段字节分别解码可能损坏一个汉字？
修改本节 lines 转换器：同时接受 LF 和 CRLF，仅在 LF 前移除一个 CR，保留空行和最后的未终止行。输入块 ["甲\r", "\n\r\n乙", "\r\n尾"] 应交付 ["甲", "", "乙", "尾"]。只接受已解码的字符串，不把任意内部 CR 全部删掉。