TransformStream.transform() 不等待 Promise 完成，导致并发执行、顺序不保；推荐用 ReadableStream + async iterator + WritableStream 实现逐块串行 await 处理。
TransformStream 本身不直接支持 async/await 的“逐块 await”，因为它的 transform() 方法必须是同步的（或返回 Promise，但浏览器目前不等待该 Promise 完成后再传入下一块）。不过，你可以通过组合 ReadableStream + WritableStream + 自定义异步逻辑，或利用 TransformStream 的 writable 和 readable 端配合 async iterator，实现真正按块顺序、可 await 的异步处理。
核心限制：TransformStream.transform() 不等 Promise
这是最容易踩坑的地方。虽然你可以写：
new TransformStream({
async transform(chunk, controller) {
const processed = await someAsyncOperation(chunk); // ✅ 语法合法
controller.enqueue(processed);
}
});但浏览器不会阻塞后续 chunk 的流入——也就是说，多个 transform() 调用可能并发执行，chunk 处理顺序不保证，也无法实现“上一块处理完，再收下一块”的流控语义。
方案一：用 ReadableStream + async iterator + WritableStream（推荐）
手动控制读取节奏：每次 await reader.read()，处理完再读下一块。天然支持串行、可 await、易错误处理。
立即学习“Java免费学习笔记（深入）”；
- 创建一个源
ReadableStream（如从 fetch、File、或自定义生成器） - 用
getReader()获取 reader，循环await reader.read() - 每拿到一块（
value），await你的异步处理逻辑 - 将结果写入目标
WritableStream（如new WritableStream({ write: ... })）
示例（模拟逐块压缩）：
async function processStreamInOrder(readable, processor) {
const reader = readable.getReader();
const writer = new WritableStream({
write(chunk) {
// 比如写入 Blob、Response.body、或另一个 stream
console.log('写入:', chunk);
}
}).getWriter();
<p>try {
while (true) {
const { done, value } = await reader.read();
if (done) break;</p><pre class="brush:php;toolbar:false;"> // ✅ 这里可以 await 任意异步操作，且严格串行
const result = await processor(value);
await writer.write(result);
}} finally { reader.releaseLock(); writer.close(); } }
// 使用 const stream = new ReadableStream({ / ... / }); await processStreamInOrder(stream, async (chunk) => { await sleep(100); // 模拟异步延迟 return new Uint8Array(chunk).map(x => x * 2); });
方案二：用 TransformStream + 内部队列 + async pump（进阶可控）
如果你仍想用 TransformStream 的标准接口（比如要链式 pipeTo），可封装一个“伪同步”转换器：内部维护一个 Promise 队列，确保 transform 提交的任务按序 resolve。
关键思路：让每个 transform() 把任务推入队列，并返回一个“排队中”的 Promise；用一个长期运行的 async function pump() 消费队列。
class AsyncTransformStream extends TransformStream {
#queue = [];
#pumpPromise = this.#startPump();
<p>constructor(transformFn, flushFn) {
super({
start: () => {},
transform: (chunk, controller) => {
this.#queue.push({ chunk, controller, transformFn });
// 返回一个暂挂 Promise，实际由 pump 控制时机
return Promise.resolve();
},
flush: flushFn ? () => flushFn() : undefined
});
}</p><p>async #startPump() {
while (this.#queue.length > 0) {
const { chunk, controller, transformFn } = this.#queue.shift();
const result = await transformFn(chunk);
controller.enqueue(result);
}
}
}</p><p>// 使用（注意：需手动 await pump 完成）
const ts = new AsyncTransformStream(
async (chunk) => {
await sleep(50);
return new TextEncoder().encode(<code>[OK]${new TextDecoder().decode(chunk)}</code>);
}
);</p><p>source.pipeThrough(ts).pipeTo(sink);
await ts.#pumpPromise; // 确保所有异步处理结束
⚠️ 注意：此方式绕过了标准流背压机制，需自行管理内存和错误传播，适合有明确生命周期的短任务。
方案三：用 streams-promise 或 web-streams-polyfill（简化开发）
第三方库如 web-streams-polyfill（含高级工具）或轻量库 streams 提供 pipeToAsync()、collect()、mapAsync() 等方法：
import { mapAsync } from 'streams';
<p>const processed = source
.pipeThrough(mapAsync(async (chunk) => {
return await heavyProcess(chunk);
}))
.getReader();</p><p>// 仍可用 async iterator 读取
for await (const chunk of processed) {
console.log(chunk);
}
这类工具底层仍是方案一的封装，但大幅降低样板代码量，适合快速落地。
