TypeScript 异步编程与迭代器
async/await 只是开始——for await of、AsyncIterator、生成器这些进阶异步能力能优雅处理流式数据。本文讲清 JS/TS 异步进阶:Promise 组合、async 迭代器(for await)、生成器、AbortController 取消、并发控制(p-limit),让异步代码既优雅又可控。
本文要点
- 主题聚焦：AI、TypeScript
- 适合人群：独立开发者、全栈工程师、AI 应用开发者
- 核心价值：快速理解背景、实践路径、常见误区和工程建议
TypeScript 异步编程与迭代器:从 async/await 走向真正可控的异步
绝大多数开发者对异步的认知止步于 async/await。写一个 fetch,前面加个 await,包一层 try/catch,完事。
但真实业务里,你会遇到更难缠的场景:大模型流式输出需要逐 token 消费;批量请求第三方接口要限速又要可取消;分页拉取直到尽头;用户切走页面要把进行中的请求全部干掉。这些都不是 await 一个 Promise 能解决的——它们要的是流、取消、并发控制这三件事。
本文把 JS/TS 异步进阶的几个关键能力一次讲透,每个都配 TS 实战代码,力求你能直接搬到生产里。
一、async/await 之后:还有更复杂的异步场景
async/await 让异步"看起来像同步",这是巨大的进步,但它解决的是单次异步操作的书写形态,而不是:
- 流式数据:数据分批到达,你想边到边处理,而不是等全部就绪
- 可取消:用户取消、超时、路由切换,你需要中断一个进行中的异步操作
- 并发控制:1000 个任务不能一起跑,但也不能串行,需要一个"池"
这三件事背后对应三个能力:AsyncIterator、AbortController、并发池。再加上 Promise 组合器和生成器,你才算真正掌握 TS 异步。
二、Promise 组合器:all / race / allSettled / any
四个组合器,语义差别经常被混淆,但只要记住一句话就够了——看它"等什么"和"失败怎么办"。
const p1 = fetch(a).then(r => r.json());
const p2 = fetch(b).then(r => r.json());
const p3 = fetch(c).then(r => r.json());
// 1) all:全部成功才算成功;任一失败立即 reject
const [a, b2, c] = await Promise.all([p1, p2, p3]);
// 2) race:谁先 settle(成功或失败)就用谁的结果,常用于超时
const data = await Promise.race([fetch(url), timeout(3000)]);
// 3) allSettled:全部完成(不管成败),返回 {status, value/reason}
const results = await Promise.allSettled([p1, p2, p3]);
results.forEach(r => {
if (r.status === 'fulfilled') console.log(r.value);
else console.error('failed', r.reason);
});
// 4) any:任一成功就成功;全部失败才 reject(AggregateError)
const fastest = await Promise.any([
fetch(mirror1).then(r => r.json()),
fetch(mirror2).then(r => r.json()),
]);
实战选择:接口聚合用 all(但要确认任何一个失败都不致命,否则会丢失其他成功结果);批量导入容忍部分失败用 allSettled,这是被低估最多的一个;多镜像抢答用 any;race 主要用来做超时竞速。
一个坑:Promise.race 不会取消另一个分支,它只是"不再等待"。真正要中断,必须配合 AbortController——下面会讲。
三、异步迭代器:for await...of 与流式数据
异步迭代器(AsyncIterable)是处理"流"的正解。任何带 Symbol.asyncIterator 的对象都能被 for await...of 消费。
最典型的例子是 Node 的 ReadableStream 或 Web Response.body:
async function consumeStream(resp: Response): Promise<string> {
const reader = resp.body!.getReader();
const decoder = new TextDecoder();
let buffer = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// 这里可以边到边处理:解析 SSE、喂给 UI 渲染
}
} finally {
reader.releaseLock();
}
return buffer;
}
但更有意思的是自己实现一个异步迭代器,比如"分页拉取直到没有下一页":
async function* paginate<T>(
fetchPage: (cursor?: string) => Promise<{ items: T[]; nextCursor?: string }>,
): AsyncGenerator<T[]> {
let cursor: string | undefined;
do {
const { items, nextCursor } = await fetchPage(cursor);
yield items;
cursor = nextCursor;
} while (cursor);
}
// 用起来极其干净
for await (const batch of paginate(api.getUsers)) {
for (const u of batch) saveUser(u);
}
注意 async function* 自动返回一个 AsyncGenerator,它天然是 AsyncIterable。SSE、WebSocket 消息流、数据库游标,都可以套这个模式。
四、生成器:function* 与 yield 的协程式表达力
生成器(function*)本身是同步的,但配合 yield* 委托、和异步生成器组合,能写出非常优雅的"惰性序列"。
// 无限自然数序列,按需消费,不占内存
function* naturals(start = 0): Generator<number> {
let i = start;
while (true) yield i++;
}
// 取前 10 个偶数
const evens = Array.from(
(function* () {
for (const n of naturals()) {
if (n > 1000) return; // 安全阀
if (n % 2 === 0) yield n;
}
})(),
).slice(0, 10);
更进阶的是用生成器做"协程式状态机"——Redux Saga 的核心思路就是 yield 出指令、由外部中间件执行副作用、再把结果塞回来。理解了这个,你就理解了"用同步代码描述异步流程"的极致形态。
但要注意:生成器不等于 async/await 的替代品,它的杀手锏是惰性和可中断的执行流。在需要按需消费、提前退出的场景(如解析器、流转换)才有不可替代价值。
五、AbortController:让异步变得可取消
AbortController 是现在 Web 平台取消异步操作的标准答案。fetch、setTimeout(实验阶段)、IndexedDB、多数 Node API 都接受一个 signal。
async function fetchWithTimeout(url: string, ms: number) {
const ctrl = new AbortController();
const timer = setTimeout(() => ctrl.abort(), ms);
try {
return await fetch(url, { signal: ctrl.signal });
} catch (e) {
if (e instanceof DOMException && e.name === 'AbortError') {
throw new Error(`请求超时 ${ms}ms`);
}
throw e;
} finally {
clearTimeout(timer);
}
}
关键概念:signal 是可传递的。一个 controller 可以同时绑定到多个 fetch、多个子任务,取消时一次性全部终止。这对"用户切走页面就停掉所有请求"非常有用:
// 在 React 组件里
useEffect(() => {
const ctrl = new AbortController();
loadInitialData(ctrl.signal);
return () => ctrl.abort(); // 组件卸载即取消
}, []);
自定义异步函数也应该接收 signal,并在循环中检查 signal.aborted——这是好库和烂库的分水岭。
六、并发控制:Promise.all 全开 vs 限并发池
Promise.all 是"全开"——所有任务同时发起。1000 个任务这么干,浏览器会排队、服务器会限流、你还可能被风控。需要限并发:同时最多 N 个在跑,空一个补一个。
工业界标准做法是 p-limit:
import pLimit from 'p-limit';
const limit = pLimit(5); // 并发上限 5
const urls = [...1000 个 URL];
const results = await Promise.all(
urls.map(url => limit(() => fetch(url).then(r => r.json()))),
);
但理解原理比会用库更重要——自己实现一个池子并不难:
async function mapPool<T, R>(
items: readonly T[],
concurrency: number,
worker: (item: T, index: number) => Promise<R>,
): Promise<R[]> {
const results: R[] = new Array(items.length);
let cursor = 0;
async function run() {
while (cursor < items.length) {
const i = cursor++;
results[i] = await worker(items[i], i);
}
}
const runners = Array.from({ length: Math.min(concurrency, items.length) }, run);
await Promise.all(runners);
return results;
}
核心思路:启动 N 个"runner",它们共享一个游标 cursor,谁有空就抢下一个任务。简单、正确、可控。
爬虫、批量 API 调用、文件批量上传,都需要这个。永远不要在生产里裸跑 Promise.all 处理大规模列表。
七、错误处理:try/catch、未捕获 rejection 与传播
async 函数里 await 抛出的错误,行为和同步 throw 一样,try/catch 能直接接住:
async function load() {
try {
const data = await fetch(url).then(r => {
if (!r.ok) throw new Error(`HTTP ${r.status}`);
return r.json();
});
return data;
} catch (e) {
// 包装成业务错误,带上上下文
throw new LoadError('加载失败', { cause: e, url });
}
}
两个高价值实践:
- 永远检查
response.ok。fetch只有网络错误才 reject,404/500 都会正常 resolve——这是无数 bug 的来源。 - 用
cause字段串错误链(ES2022)。保留原始堆栈,方便排障。
另一个常被忽略的点:未捕获的 Promise rejection。在 Node 里,默认会让进程崩溃(unhandledRejection)。养成两个习惯:链路上每个 async 函数都明确处理或向上抛;监听全局兜底:
process.on('unhandledRejection', (reason) => {
logger.error('未捕获 rejection', reason);
});
Promise.all 里任一失败会丢失其他子任务的成功结果——如果不想丢,用 allSettled。
八、实战:流式读取 + 限并发 + 可取消
把上面几个能力组合起来,就是一个真实场景:从大模型流式拉 token,同时对历史会话做限并发批量向量化,而且整个流程可被用户取消。
interface ChatChunk { delta: string; done: boolean }
async function* streamChat(
prompt: string,
signal: AbortSignal,
): AsyncGenerator<string> {
const resp = await fetch('/api/chat', {
method: 'POST',
body: JSON.stringify({ prompt }),
signal,
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const reader = resp.body!.getReader();
const decoder = new TextDecoder();
let buf = '';
try {
while (true) {
if (signal.aborted) throw new DOMException('Canceled', 'AbortError');
const { done, value } = await reader.read();
if (done) break;
buf += decoder.decode(value, { stream: true });
// SSE 协议:按 \n\n 切分事件
const parts = buf.split('\n\n');
buf = parts.pop() ?? '';
for (const part of parts) {
const chunk: ChatChunk = JSON.parse(part.replace(/^data: /, ''));
if (chunk.delta) yield chunk.delta;
}
}
} finally {
reader.releaseLock();
}
}
async function vectorizeAll(
texts: string[],
signal: AbortSignal,
): Promise<number[][]> {
return mapPool(texts, 8, async (text) => {
const resp = await fetch('/api/embed', {
method: 'POST',
body: JSON.stringify({ text }),
signal, // 共享同一个 signal,取消时全部停
});
return (await resp.json()).vector as number[];
});
}
// 主流程
async function run(prompt: string, history: string[]) {
const ctrl = new AbortController();
// 5 秒后自动取消(演示)
setTimeout(() => ctrl.abort(), 5000);
// 流式输出 + 并行向量化,共享一个 AbortController
const [vectors, tokens] = await Promise.all([
vectorizeAll(history, ctrl.signal).catch(() => []),
(async () => {
let full = '';
try {
for await (const t of streamChat(prompt, ctrl.signal)) {
full += t;
process.stdout.write(t); // 实时渲染到 UI
}
} catch (e) {
if (!(e instanceof DOMException)) throw e;
}
return full;
})(),
]);
return { vectors, tokens };
}
这段代码集中体现了文章开头的三个关键词:流(异步生成器消费 SSE)、取消(共享 AbortSignal 一次性终止 fetch 和池)、并发控制(mapPool 限并发 8)。把这套结构抽象好,几乎所有 AI 应用的数据层都能复用。
小结
- async/await 只是书写形态,真正难的是流、取消、并发——三者背后分别是 AsyncIterator、AbortController、并发池。
- Promise 组合器按"等什么 + 失败怎么办"选:
all全成功、allSettled全完成(最被低估)、any任一成功、race谁先到。 - 异步迭代器是流的正解:
async function*+for await...of,SSE、分页、读流都能写得像同步循环一样优雅。 - AbortController 是取消的唯一标准答案,
signal可传递、可共享;自定义异步函数暴露signal参数是好库的底线。 - 大规模任务必须限并发:
p-limit拿来就用,但更推荐自己写一遍池子——简单的游标 + N 个 runner,十行代码,一辈子受用。
一句话金句:async/await 让异步可读,而 AsyncIterator、AbortController、并发池让异步可控。可读是及格,可控才是工程师的成年礼。
Related