掌握async/await的进阶技巧,让你的异步代码更加优雅、健壮、高效。本文深入探讨async生成器、取消机制、并发控制等高级特性,帮助架构师和中高级开发者写出生产级的异步代码。
目录
为什么需要高级异步模式
如果你已经熟练掌握了 async/await 的基本用法:
async function fetchUser(id) {
try {
const response = await fetch(`/api/users/${id}`);
const user = await response.json();
return user;
} catch (error) {
console.error('获取用户失败:', error);
throw error;
}
}
那么恭喜你,你已经能够处理80%的异步场景了。但在生产环境中,你会遇到更棘手的问题:
-
分页数据流: 如何优雅地处理无限滚动的分页数据?
-
请求取消: 用户离开页面时,如何取消正在进行的请求?
-
并发控制: 100个并发请求会打爆服务器,如何控制到最多5个?
-
错误隔离: 一个请求失败不应该拖垮整个应用,如何实现错误边界?
-
资源清理: 如何确保文件句柄、数据库连接等资源被正确释放?
这些问题需要高级异步模式来解决。本文将带你深入这5个核心领域,让你写出像V8引擎源码一样优雅的异步代码。
Async Generator: 异步迭代器
什么是异步生成器
异步生成器是 ES2018 引入的特性,它结合了三个概念:
-
异步(async): 支持await操作
-
生成器(generator): 可以暂停执行,惰性求值
-
迭代器(iterator): 可以用for...of遍历
简单来说,异步生成器让你可以一边异步获取数据,一边逐个产出结果,而不是等所有数据加载完才返回。
基础语法
// 定义异步生成器函数
async function* asyncGenerator() {
yield Promise.resolve(1);
yield Promise.resolve(2);
yield Promise.resolve(3);
}
// 消费异步生成器
async function consume() {
for await (const value of asyncGenerator()) {
console.log(value); // 1, 2, 3
}
}
关键点:
- 使用
async function*定义
-
yield可以产出Promise或普通值
- 必须用
for await...of来消费
- 每个
yield都会等待前面的异步操作完成
实际应用场景
场景1: 分页API数据流
假设你有一个分页API,需要获取所有数据:
// 传统方式:一次性加载所有数据
async function fetchAllPages(url) {
const allData = [];
let page = 1;
while (true) {
const response = await fetch(`${url}?page=${page}`);
const data = await response.json();
if (data.items.length === 0) break;
allData.push(...data.items);
page++;
}
return allData; // 可能几千条数据,内存爆炸!
}
// 异步生成器:流式处理
async function* fetchPaginatedData(url) {
let page = 1;
while (true) {
const response = await fetch(`${url}?page=${page}`);
const data = await response.json();
if (data.items.length === 0) break;
yield data; // 每次只产出当前页数据
page++;
}
}
// 消费:内存友好,处理一页释放一页
async function processLargeDataset() {
for await (const page of fetchPaginatedData('/api/users')) {
await processPage(page.items); // 处理当前页
// 处理完成后,当前页数据可被垃圾回收
}
}
优势:
- 内存占用恒定,不随数据量增长
- 真正的流式处理,边加载边处理
- 更早开始处理,不用等所有数据加载完
场景2: 文件逐行读取
async function* readLines(filepath) {
const fileHandle = await fs.promises.open(filepath, 'r');
const buffer = Buffer.alloc(1024);
let leftover = '';
try {
while (true) {
const { bytesRead } = await fileHandle.read(buffer, 0, 1024);
if (bytesRead === 0) {
if (leftover) yield leftover;
break;
}
const chunk = leftover + buffer.toString('utf8', 0, bytesRead);
const lines = chunk.split('\n');
// 最后一行可能不完整,留到下次处理
leftover = lines.pop() || '';
for (const line of lines) {
yield line;
}
}
} finally {
await fileHandle.close(); // 确保文件句柄被释放
}
}
// 使用
async function processLogFile() {
for await (const line of readLines('/var/log/app.log')) {
if (line.includes('ERROR')) {
console.error(line);
}
}
}
关键点: 使用try/finally确保资源清理,这是异步生成器的最佳实践。
场景3: 实时数据流
假设你在对接WebSocket实时数据流:
async function* streamMessages(websocketUrl) {
const ws = new WebSocket(websocketUrl);
const messageQueue = [];
let resolveMessage = null;
ws.onmessage = (event) => {
if (resolveMessage) {
resolveMessage(JSON.parse(event.data));
resolveMessage = null;
} else {
messageQueue.push(JSON.parse(event.data));
}
};
try {
while (true) {
if (messageQueue.length > 0) {
yield messageQueue.shift();
} else {
yield await new Promise(resolve => {
resolveMessage = resolve;
});
}
}
} finally {
ws.close(); // 确保WebSocket连接关闭
}
}
// 消费实时流
async function monitorRealtimeData() {
const abortController = new AbortController();
setTimeout(() => abortController.abort(), 60000); // 1分钟后停止
for await (const message of streamMessages('wss://api.example.com/realtime')) {
if (abortController.signal.aborted) break;
await processMessage(message);
}
}
高级技巧
1. yield* 委托
async function* userPosts(userId) {
const posts = await fetchUserPosts(userId);
yield* posts; // 委托给数组的同步迭代器
}
async function* allUsersPosts(userIds) {
for (const userId of userIds) {
yield* userPosts(userId); // 委托给另一个异步生成器
}
}
// 平铺所有用户的文章
for await (const post of allUsersPosts([1, 2, 3])) {
console.log(post.title);
}
2. 从异步生成器创建ReadableStream
async function* generateChunks() {
yield 'chunk1';
yield 'chunk2';
yield 'chunk3';
}
const stream = new ReadableStream({
async start(controller) {
for await (const chunk of generateChunks()) {
controller.enqueue(chunk);
}
controller.close();
}
});
// 可以用于fetch的body
await fetch('/api/upload', {
method: 'POST',
body: stream
});
AbortController: 优雅的取消机制
为什么需要取消机制
想象一个场景:用户在搜索框输入,每次输入都会触发API请求。如果用户快速输入"JavaScript",会连续触发10个请求,但只有最后一个请求是有意义的,前9个都是浪费:
// ❌ 问题代码
let currentQuery = '';
async function handleSearch(query) {
currentQuery = query;
const results = await fetch(`/api/search?q=${query}`);
if (query === currentQuery) { // 竞态条件检查
displayResults(results);
}
}
问题:
- 前9个请求仍在执行,浪费带宽
- 无法真正中断请求,只能忽略结果
- 代码逻辑复杂,容易出现竞态条件
解决方案: AbortController
AbortController基础
const controller = new AbortController();
const signal = controller.signal;
// 发起可取消的请求
fetch('/api/data', { signal })
.then(response => response.json())
.then(data => console.log(data))
.catch(error => {
if (error.name === 'AbortError') {
console.log('请求被取消');
} else {
console.error('请求失败:', error);
}
});
// 取消请求
controller.abort();
核心API:
-
controller.abort(reason?): 取消所有关联的操作
-
controller.signal: 一个AbortSignal对象,可以传递给异步操作
-
signal.aborted: 布尔值,表示是否已取消
-
signal.reason: 取消的原因
-
signal.throwIfAborted(): 如果已取消,则抛出错误
实际应用场景
场景1: 搜索请求取消
let abortController = null;
async function search(query) {
// 取消前一个请求
if (abortController) {
abortController.abort();
}
// 创建新的控制器
abortController = new AbortController();
try {
const response = await fetch(`/api/search?q=${query}`, {
signal: abortController.signal
});
const results = await response.json();
return results;
} catch (error) {
if (error.name === 'AbortError') {
console.log('搜索请求被取消');
return [];
}
throw error;
}
}
// 用户快速输入: J -> Ja -> Jav -> Java -> JavaS -> ...
// 只有最后一个请求会执行,前面的都被取消
场景2: React组件卸载清理
import { useEffect, useState } from 'react';
function UserProfile({ userId }) {
const [user, setUser] = useState(null);
const [loading, setLoading] = useState(true);
useEffect(() => {
const abortController = new AbortController();
async function fetchUser() {
try {
setLoading(true);
const response = await fetch(`/api/users/${userId}`, {
signal: abortController.signal
});
const data = await response.json();
setUser(data);
} catch (error) {
if (error.name !== 'AbortError') {
console.error('获取用户失败:', error);
}
} finally {
setLoading(false);
}
}
fetchUser();
// 组件卸载时取消请求
return () => abortController.abort();
}, [userId]);
if (loading) return <div>Loading...</div>;
return <div>{user?.name}</div>;
}
关键点: useEffect的清理函数会在组件卸载时或userId变化时执行,确保旧请求被取消。
场景3: 超时控制
现代浏览器支持直接创建超时signal:
// 方式1: AbortSignal.timeout (现代浏览器)
async function fetchWithTimeout(url, timeout = 5000) {
try {
const response = await fetch(url, {
signal: AbortSignal.timeout(timeout)
});
return response.json();
} catch (error) {
if (error.name === 'AbortError' || error.name === 'TimeoutError') {
throw new Error(`请求超时 (${timeout}ms)`);
}
throw error;
}
}
// 方式2: 手动超时控制 (兼容性更好)
async function fetchWithManualTimeout(url, timeout = 5000) {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeout);
try {
const response = await fetch(url, {
signal: controller.signal
});
return response.json();
} finally {
clearTimeout(timeoutId);
}
}
场景4: 取消多个操作
AbortController可以同时取消多个操作:
async function fetchDashboardData(userId) {
const controller = new AbortController();
const signal = controller.signal;
try {
// 同时发起多个请求,共享同一个signal
const [user, posts, comments] = await Promise.all([
fetch(`/api/users/${userId}`, { signal }).then(r => r.json()),
fetch(`/api/users/${userId}/posts`, { signal }).then(r => r.json()),
fetch(`/api/users/${userId}/comments`, { signal }).then(r => r.json())
]);
return { user, posts, comments };
} catch (error) {
if (error.name === 'AbortError') {
console.log('Dashboard数据请求被取消');
return null;
}
throw error;
}
}
// 用户离开页面时取消所有请求
window.addEventListener('beforeunload', () => {
controller.abort();
});
自定义可取消操作
AbortController不仅限于fetch,你可以为任何异步操作添加取消功能:
// 可取消的延时
function cancellableDelay(ms, signal) {
return new Promise((resolve, reject) => {
const timeoutId = setTimeout(resolve, ms);
signal.addEventListener('abort', () => {
clearTimeout(timeoutId);
reject(new DOMException('Aborted', 'AbortError'));
}, { once: true }); // once: true 确保只触发一次
});
}
// 使用
async function countdown(seconds, signal) {
for (let i = seconds; i > 0; i--) {
console.log(i);
await cancellableDelay(1000, signal);
}
console.log('完成!');
}
const controller = new AbortController();
countdown(10, controller.signal).catch(console.error);
// 3秒后取消
setTimeout(() => controller.abort(), 3000);
// 输出: 10, 9, 8, 然后抛出AbortError
并发控制: 从失控到可控
为什么需要并发控制
Promise.all()虽然方便,但没有并发限制:
// ❌ 危险代码:可能同时发起1000个请求!
const userIds = Array.from({ length: 1000 }, (_, i) => i + 1);
await Promise.all(
userIds.map(id => fetch(`/api/users/${id}`))
);
// 服务器会被瞬间打爆,触发限流,甚至被拉黑
现实问题:
- 浏览器对同一域名的并发请求有限制(通常6个)
- 服务器有速率限制(rate limit)
- 数据库连接池有限
- 内存占用会飙升
并发控制方案对比
| 方案 |
特点 |
npm包 |
周下载量 |
| Promise.all |
无限制并发 |
原生 |
- |
| p-limit |
简单限流 |
p-limit |
2.04亿 |
| p-map |
映射处理 |
p-map |
6780万 |
| p-queue |
队列管理 |
p-queue |
1890万 |
方案1: p-limit (推荐)
最简单轻量的并发控制库:
npm install p-limit
import pLimit from 'p-limit';
// 创建并发限制器(最多同时执行3个任务)
const limit = pLimit(3);
async function fetchUsers() {
const userIds = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
// 将每个任务包装到limit()中
const tasks = userIds.map(id =>
limit(() => fetch(`/api/users/${id}`).then(r => r.json()))
);
// 执行所有任务,但最多同时执行3个
const users = await Promise.all(tasks);
console.log('获取用户:', users.length); // 10
return users;
}
工作原理:
- pLimit(3)创建一个限制器
- limit(fn)将任务加入队列
- 如果当前执行数<3,立即执行
- 如果当前执行数>=3,等待前面的任务完成
- 任务完成后,从队列中取出下一个执行
方案2: p-map
专门用于映射处理的并发控制:
npm install p-map
import pMap from 'p-map';
async function processImages(imageUrls) {
const results = await pMap(
imageUrls,
async (url) => {
const response = await fetch(url);
const blob = await response.blob();
return processImage(blob);
},
{
concurrency: 5, // 最多同时处理5张图片
stopOnError: false // 即使有错误也继续处理其他图片
}
);
return results;
}
高级选项:
-
concurrency: 并发数
-
stopOnError: 是否在遇到错误时停止(默认true)
-
signal: AbortSignal,支持取消
方案3: p-queue
功能最全面的队列管理:
npm install p-queue
import PQueue from 'p-queue';
// 创建队列
const queue = new PQueue({
concurrency: 3, // 最多同时执行3个任务
interval: 1000, // 每秒最多开始3个任务
intervalCap: 3, // 每个interval期间最多执行的任务数
});
// 监听事件
queue.on('active', () => {
console.log(`正在执行: ${queue.pending}/${queue.concurrency}`);
});
queue.on('idle', () => {
console.log('队列为空,所有任务完成');
});
queue.on('error', (error) => {
console.error('任务执行出错:', error);
});
// 添加任务
async function fetchUserData() {
for (let i = 1; i <= 100; i++) {
queue.add(() =>
fetch(`/api/users/${i}`)
.then(r => r.json())
.then(user => {
console.log(`获取用户: ${user.name}`);
return user;
})
);
}
// 等待所有任务完成
await queue.onIdle();
console.log('所有用户获取完成');
}
高级特性:
-
优先级:
queue.add(fn, { priority: 10 })
-
暂停/恢复:
queue.pause() / queue.start()
-
清空队列:
queue.clear()
-
统计信息:
queue.pending, queue.size
方案4: 手动实现并发控制
如果你不想引入依赖,可以手动实现:
async function limitConcurrency(tasks, concurrency) {
const results = [];
const executing = new Set();
for (const [index, task] of tasks.entries()) {
const promise = task().then(result => {
executing.delete(promise);
results[index] = result;
return result;
});
executing.add(promise);
if (executing.size >= concurrency) {
await Promise.race(executing);
}
}
return Promise.all(executing).then(() => results);
}
// 使用
const tasks = userIds.map(id => () => fetch(`/api/users/${id}`).then(r => r.json()));
const users = await limitConcurrency(tasks, 3);
核心算法:
- 维护一个执行中的Promise集合
executing
- 每次添加新任务时,检查执行数是否达到上限
- 达到上限时,等待任意一个任务完成(
Promise.race)
- 任务完成后自动从集合中移除,空出位置给下一个任务
并发控制最佳实践
1. 根据资源设置并发数
// 根据CPU核心数
import os from 'os';
const cpuCount = os.cpus().length;
const limit = pLimit(cpuCount);
// 根据数据库连接池
const dbPoolSize = 10;
const dbLimit = pLimit(dbPoolSize - 2); // 留2个备用
// 根据API速率限制
const rateLimitPerSecond = 50;
const apiLimit = pLimit(rateLimitPerSecond / 2); // 留buffer
2. 动态调整并发数
const queue = new PQueue({ concurrency: 5 });
// 根据响应时间动态调整
setInterval(() => {
const avgResponseTime = calculateAverageResponseTime();
if (avgResponseTime > 2000) {
// 响应慢,减少并发
queue.concurrency = Math.max(1, queue.concurrency - 1);
} else if (avgResponseTime < 500) {
// 响应快,增加并发
queue.concurrency = Math.min(20, queue.concurrency + 1);
}
}, 10000);
3. 错误处理
async function fetchWithErrorHandling(items) {
const errors = [];
const results = await pMap(
items,
async (item) => {
try {
return await processItem(item);
} catch (error) {
errors.push({ item, error });
return null; // 返回null而不是抛出错误
}
},
{
concurrency: 5,
stopOnError: false
}
);
if (errors.length > 0) {
console.error(`处理失败: ${errors.length}/${items.length}`);
errors.forEach(({ item, error }) => {
console.error(`项目${item.id}处理失败:`, error);
});
}
return results.filter(r => r !== null);
}
高级错误处理模式
Try-Catch的局限性
传统的try-catch在复杂场景中不够用:
async function processOrder(orderId) {
try {
const order = await fetchOrder(orderId);
const inventory = await checkInventory(order.items);
const payment = await processPayment(order);
const shipment = await createShipment(order);
return { order, inventory, payment, shipment };
} catch (error) {
// 不知道是哪一步出错
console.error('处理订单失败:', error);
throw error;
}
}
问题:
- 无法区分哪一步出错
- 嵌套try-catch导致代码膨胀
- 错误吞噬(catch后未处理)
- 无法针对不同错误类型采取不同策略
模式1: 错误分类
// 定义自定义错误类型
class NetworkError extends Error {
constructor(message, statusCode) {
super(message);
this.name = 'NetworkError';
this.statusCode = statusCode;
}
}
class ValidationError extends Error {
constructor(message, fields) {
super(message);
this.name = 'ValidationError';
this.fields = fields;
}
}
class PaymentError extends Error {
constructor(message, orderId) {
super(message);
this.name = 'PaymentError';
this.orderId = orderId;
}
}
// 使用
async function processOrder(orderId) {
try {
const order = await fetchOrder(orderId);
// 验证错误
if (!order.items || order.items.length === 0) {
throw new ValidationError('订单没有商品', ['items']);
}
// 网络错误
const inventory = await checkInventory(order.items);
if (inventory.status === 503) {
throw new NetworkError('库存服务不可用', 503);
}
// 支付错误
const payment = await processPayment(order);
if (payment.status === 'failed') {
throw new PaymentError('支付失败', orderId);
}
return { order, inventory, payment };
} catch (error) {
// 根据错误类型采取不同策略
if (error instanceof ValidationError) {
return { success: false, reason: 'validation', fields: error.fields };
} else if (error instanceof NetworkError) {
// 重试逻辑
return retryOperation(() => processOrder(orderId), 3);
} else if (error instanceof PaymentError) {
// 发送告警
alertSupport(error.orderId, error.message);
return { success: false, reason: 'payment' };
} else {
throw error; // 未知错误,向上抛出
}
}
}
模式2: 结果包装器
// Result类型 (类似Rust的Result)
class Result {
static ok(value) {
return { success: true, value };
}
static error(error) {
return { success: false, error };
}
}
async function safeAsync(promise) {
try {
const value = await promise;
return Result.ok(value);
} catch (error) {
return Result.error(error);
}
}
// 使用
async function processOrders(orderIds) {
const results = await Promise.all(
orderIds.map(id => safeAsync(processOrder(id)))
);
const successful = results.filter(r => r.success).map(r => r.value);
const failed = results.filter(r => !r.success).map(r => r.error);
console.log(`成功: ${successful.length}, 失败: ${failed.length}`);
// 失败的错误可以记录日志、重试或告警
failed.forEach(error => {
console.error('订单处理失败:', error);
});
return { successful, failed };
}
模式3: 错误包装器
特别适用于Express.js等框架:
// Express错误包装器
function asyncHandler(fn) {
return (req, res, next) => {
Promise.resolve(fn(req, res, next)).catch(next);
};
}
// 使用
app.get('/users/:id', asyncHandler(async (req, res) => {
const user = await fetchUser(req.params.id);
res.json(user);
}));
// 全局错误处理中间件
app.use((error, req, res, next) => {
if (error instanceof ValidationError) {
res.status(400).json({ error: 'validation', fields: error.fields });
} else if (error instanceof NetworkError) {
res.status(502).json({ error: 'network', message: error.message });
} else {
console.error('未处理的错误:', error);
res.status(500).json({ error: 'internal' });
}
});
模式4: Promise.allSettled并发错误处理
当多个并发任务且不想一个失败导致全部失败时:
async function fetchAllData(userId) {
const results = await Promise.allSettled([
fetch(`/api/users/${userId}`).then(r => r.json()),
fetch(`/api/users/${userId}/posts`).then(r => r.json()),
fetch(`/api/users/${userId}/comments`).then(r => r.json()),
fetch(`/api/users/${userId}/followers`).then(r => r.json()),
]);
const [user, posts, comments, followers] = results.map((result, index) => {
if (result.status === 'fulfilled') {
return result.value;
} else {
console.error(`数据${index}获取失败:`, result.reason);
return null; // 返回null作为fallback
}
});
return {
user: user || {},
posts: posts || [],
comments: comments || [],
followers: followers || [],
};
}
模式5: 全局错误捕获
作为最后一道防线,捕获未处理的异常:
// Node.js
process.on('unhandledRejection', (reason, promise) => {
console.error('未处理的Promise rejection:', reason);
// 发送到错误追踪服务
Sentry.captureException(reason);
process.exit(1); // 可选:强制退出进程
});
process.on('uncaughtException', (error) => {
console.error('未捕获的异常:', error);
Sentry.captureException(error);
process.exit(1);
});
// 浏览器
window.addEventListener('unhandledrejection', (event) => {
console.error('未处理的Promise rejection:', event.reason);
event.preventDefault(); // 阻止默认行为(控制台错误)
});
window.addEventListener('error', (event) => {
console.error('全局错误:', event.error);
});
// React Error Boundary
import { ErrorBoundary } from 'react-error-boundary';
function ErrorFallback({ error, resetErrorBoundary }) {
return (
<div role="alert">
<p>出错了:</p>
<pre>{error.message}</pre>
<button onClick={resetErrorBoundary}>重试</button>
</div>
);
}
<ErrorBoundary FallbackComponent={ErrorFallback}>
<App />
</ErrorBoundary>
实战案例:分页API完整实现
让我们把学到的知识整合起来,实现一个生产级的分页数据获取方案:
/**
* 分页API客户端
* 支持:
* - 异步生成器流式处理
* - AbortController取消
* - 并发控制
* - 错误处理
*/
class PaginatedAPIClient {
constructor(baseUrl, options = {}) {
this.baseUrl = baseUrl;
this.concurrency = options.concurrency || 5;
this.timeout = options.timeout || 30000;
this.cache = new Map();
}
/**
* 异步生成器:流式获取分页数据
*/
async *fetchPaginated(endpoint, options = {}) {
const signal = options.signal || new AbortController().signal;
let page = options.startPage || 1;
let hasMore = true;
while (hasMore && !signal.aborted) {
const url = `${this.baseUrl}${endpoint}?page=${page}&${new URLSearchParams(options.params || {})}`;
try {
const response = await this.fetchWithTimeout(url, signal);
const data = await response.json();
if (!data.items || data.items.length === 0) {
hasMore = false;
break;
}
yield data;
hasMore = data.hasMore !== false;
page++;
} catch (error) {
if (error.name === 'AbortError') {
console.log('请求被取消');
break;
}
throw error;
}
}
}
/**
* 批量处理分页数据(支持并发)
*/
async processPaginated(endpoint, processor, options = {}) {
const results = [];
const errors = [];
const batchSize = options.batchSize || 10;
const signal = options.signal || new AbortController().signal;
let batch = [];
for await (const page of this.fetchPaginated(endpoint, options)) {
if (signal.aborted) break;
batch.push(page);
if (batch.length >= batchSize) {
const batchResults = await this.processBatch(batch, processor, signal);
results.push(...batchResults.successful);
errors.push(...batchResults.errors);
batch = [];
}
}
// 处理剩余批次
if (batch.length > 0) {
const batchResults = await this.processBatch(batch, processor, signal);
results.push(...batchResults.successful);
errors.push(...batchResults.errors);
}
return { results, errors };
}
/**
* 批量处理(内部方法,支持并发控制)
*/
async processBatch(pages, processor, signal) {
const results = await Promise.allSettled(
pages.map(page => processor(page, signal))
);
return {
successful: results
.filter(r => r.status === 'fulfilled')
.map(r => r.value),
errors: results
.filter(r => r.status === 'rejected')
.map(r => r.reason)
};
}
/**
* 带超时的fetch
*/
async fetchWithTimeout(url, signal) {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), this.timeout);
// 合并外部signal和timeout signal
if (signal) {
signal.addEventListener('abort', () => controller.abort());
}
try {
const response = await fetch(url, { signal: controller.signal });
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
return response;
} finally {
clearTimeout(timeoutId);
}
}
/**
* 带缓存的get请求
*/
async get(endpoint, options = {}) {
const cacheKey = `${endpoint}:${JSON.stringify(options.params)}`;
if (options.useCache && this.cache.has(cacheKey)) {
return this.cache.get(cacheKey);
}
const url = `${this.baseUrl}${endpoint}?${new URLSearchParams(options.params || {})}`;
const response = await this.fetchWithTimeout(url, options.signal);
const data = await response.json();
if (options.useCache) {
this.cache.set(cacheKey, data);
}
return data;
}
/**
* 取消所有正在进行的请求
*/
cancelAll() {
this.activeRequests.forEach(controller => controller.abort());
this.activeRequests.clear();
}
}
// 使用示例
async function main() {
const client = new PaginatedAPIClient('https://api.example.com', {
concurrency: 5,
timeout: 15000
});
// 示例1: 流式处理大数据集
const abortController = new AbortController();
setTimeout(() => {
console.log('30秒后取消请求');
abortController.abort();
}, 30000);
for await (const page of client.fetchPaginated('/users', {
signal: abortController.signal,
params: { status: 'active' }
})) {
console.log(`处理第${page.currentPage}页,共${page.items.length}条数据`);
await processUserData(page.items);
}
// 示例2: 批量处理(带并发控制)
const { results, errors } = await client.processPaginated(
'/posts',
async (page, signal) => {
// 每页数据的处理逻辑
const processedItems = [];
for (const post of page.items) {
if (signal.aborted) break;
// 处理post
const enriched = await enrichPost(post);
processedItems.push(enriched);
}
return processedItems;
},
{
batchSize: 10,
startPage: 1,
signal: abortController.signal
}
);
console.log(`成功: ${results.length}, 失败: ${errors.length}`);
}
// 启动
main().catch(console.error);
这个实现整合了:
-
异步生成器: 流式处理分页数据
-
AbortController: 支持取消请求
-
并发控制: 批量处理时限制并发
-
错误处理: Promise.allSettled捕获错误
-
超时控制: 防止请求hang住
-
缓存: 减少重复请求
-
类型安全: 可配合TypeScript使用
性能优化建议
1. 避免async/await性能陷阱
陷阱1: 不必要的await
// ❌ 低效:串行执行两个不相关的操作
async function getData() {
const user = await fetchUser();
const posts = await fetchPosts();
return { user, posts };
}
// ✅ 高效:并行执行
async function getData() {
const [user, posts] = await Promise.all([
fetchUser(),
fetchPosts()
]);
return { user, posts };
}
陷阱2: 循环中的await
// ❌ 低效:串行执行
async function fetchUsers(userIds) {
const users = [];
for (const id of userIds) {
const user = await fetch(`/api/users/${id}`).then(r => r.json());
users.push(user);
}
return users;
}
// ✅ 高效:并发控制
async function fetchUsers(userIds) {
const limit = pLimit(5); // 限制并发数为5
const tasks = userIds.map(id =>
limit(() => fetch(`/api/users/${id}`).then(r => r.json()))
);
return Promise.all(tasks);
}
2. 内存优化
// ❌ 内存占用高
async function fetchAllData() {
const allData = [];
for await (const page of fetchPages()) {
allData.push(...page.items); // 数据越来越多,内存爆炸
}
return allData;
}
// ✅ 流式处理,内存友好
async function processAllData() {
for await (const page of fetchPages()) {
await processItems(page.items); // 处理完就释放
}
}
3. 超时和重试策略
async function fetchWithRetry(url, options = {}) {
const { retries = 3, timeout = 5000 } = options;
for (let i = 0; i < retries; i++) {
try {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeout);
const response = await fetch(url, {
...options,
signal: controller.signal
});
clearTimeout(timeoutId);
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
return response.json();
} catch (error) {
if (error.name === 'AbortError') {
console.log(`请求超时,第${i + 1}次重试`);
} else {
console.error(`请求失败(${error.message}),第${i + 1}次重试`);
}
if (i === retries - 1) {
throw error;
}
// 指数退避
await new Promise(resolve =>
setTimeout(resolve, Math.pow(2, i) * 1000)
);
}
}
}
4. 使用性能监控工具
// 测量异步函数执行时间
async function measurePerformance(name, fn) {
const start = performance.now();
const result = await fn();
const duration = performance.now() - start;
console.log(`${name}耗时: ${duration.toFixed(2)}ms`);
return result;
}
// 使用
await measurePerformance('fetchUsers', () => fetchUsers([1, 2, 3]));
总结
本文深入探讨了async/await的5个高级使用模式:
核心要点回顾
-
Async Generator
- 流式处理大数据集
- 惰性求值,内存友好
- 使用
for await...of消费
- 用
try/finally确保资源清理
-
AbortController
- 优雅取消异步操作
- 组件卸载时清理请求
- 避免竞态条件
- 支持超时控制
-
并发控制
- Promise.all无限并发,危险!
- p-limit简单轻量
- p-map适合映射处理
- p-queue功能最全
-
错误处理
- 区分错误类型
- Promise.allSettled并发容错
- 全局错误捕获作为防线
- React Error Boundary
-
性能优化
- 避免不必要的await
- 使用并发控制代替串行
- 流式处理减少内存占用
- 添加超时和重试
知识检查清单
- 能用async生成器处理分页API
- 能用AbortController取消fetch请求
- 能选择合适的并发控制库(p-limit/p-map/p-queue)
- 能区分不同的错误处理场景并选择合适策略
- 能实现生产级的分页API客户端
进阶学习路线
-
源码阅读: 阅读p-limit、p-queue的源码(都很小,易于理解)
-
实践项目: 用async生成器实现一个文件流处理器
-
深入原理: 了解V8引擎的async/await实现
-
性能对比: Benchmark不同并发控制方案的性能差异
参考链接
登录
