Python 异步编程实战用 asyncio 让你的网络请求提速 7 倍前言在 Web 全栈开发中网络请求是不可避免的性能瓶颈。假设你需要从 10 个 API 接口拉取数据每个接口耗时约 1 秒——同步方式需要 10 秒而使用 Python 的asyncio异步编程只需要 1 秒多。本文将通过一个完整的实战案例带你从零掌握 Python 异步编程的核心概念和最佳实践。一、为什么需要异步编程1.1 三种并发模型Python 提供了三种并发编程方式模型适用场景是否受 GIL 限制threading多线程I/O 密集型✅ 是multiprocessing多进程CPU 密集型❌ 否asyncio异步协程网络 I/O 密集型✅ 是但无影响关键认知网络请求的大部分时间都在等待响应CPU 几乎空闲。异步编程就是利用这段等待时间去处理其他请求——而不是傻等。1.2 同步 vs 异步一张图看懂同步顺序执行 请求1 ████████░░░░ (等待) ████████ 请求2 ████████░░░░ (等待) ████████ 请求3 ████████░░░░... 总耗时 ≈ 10 秒 异步并发执行 请求1 ████████░░░░ (等待) ████████ 请求2 ████████░░░░ (等待) ████████ 请求3 ████████░░░░ (等待) ████████ 总耗时 ≈ 1.5 秒 ← 几乎等于最慢那个请求的时间二、核心概念速览在写代码之前先理解几个关键概念2.1async def和await# 普通函数defnormal_func():returnhello# 协程函数async defasyncdefasync_func():awaitasyncio.sleep(1)# 模拟异步等待returnhelloasync def定义一个协程函数调用它不会立即执行而是返回一个协程对象await暂停当前协程把控制权交还给事件循环让其他协程先跑如果你有 JavaScript 经验Python 的async/await和 JS 的async/await概念几乎一样只是运行环境从浏览器的事件循环换成了 Python 的asyncio事件循环。2.2 事件循环Event Loop事件循环是异步编程的调度中心——它维护一个任务队列哪个协程在等待 I/O就切换到另一个协程执行。importasyncioasyncdefmain():resultawaitasync_func()# Python 3.7 的标准入口asyncio.run(main())三、实战并发抓取 10 个 URL3.1 模拟场景我们写一个fetch_url函数来模拟网络请求importasyncioimportrandomasyncdeffetch_url(url:str)-str:模拟网络请求延迟 0.5~1.5 秒10% 概率失败delayrandom.uniform(0.5,1.5)awaitasyncio.sleep(delay)ifrandom.random()0.1:raiseConnectionError(f请求超时:{url})returnf{url} 响应 (耗时{delay:.2f}s)3.2 ❌ 错误示范顺序执行asyncdefsequential_fetch(urls):results[]forurlinurls:resultawaitfetch_url(url)# 一个接一个等results.append(result)returnresults# 耗时: ~10 秒这就是披着 async 外衣的同步代码——for循环里逐个await本质上还是顺序执行。3.3 ✅ 正确姿势asyncio.gather wrapper 模式asyncdefasync_fetch_all(urls:list[str])-tuple[list[str],list[str]]:并发抓取所有 URL单个失败不影响其他# Wrapper在协程内部捕获异常避免一个失败导致全部崩溃asyncdefsafe_fetch(url:str):try:resultawaitfetch_url(url)return(ok,result,url)exceptConnectionError:return(fail,None,url)# 一次性发起所有请求resultsawaitasyncio.gather(*[safe_fetch(url)forurlinurls])# 拆分成功/失败success[]failed[]forstatus,result,urlinresults:ifstatusok:success.append(result)else:failed.append(url)returnsuccess,failed3.4 性能对比importtime# 同步版本starttime.perf_counter()asyncio.run(sequential_fetch(test_urls))sync_timetime.perf_counter()-startprint(f同步耗时:{sync_time:.2f}s)# 异步版本starttime.perf_counter()success,failedasyncio.run(async_fetch_all(test_urls))async_timetime.perf_counter()-startprint(f异步耗时:{async_time:.2f}s)print(f提速:{sync_time/async_time:.1f}x)实际运行结果同步耗时: 10.52s 异步耗时: 1.37s 提速: 7.7x ← 10 个请求几乎同时完成四、关键模式Wrapper 模式详解4.1 为什么需要 Wrapperasyncio.gather默认行为如果其中任何一个协程抛出异常整个 gather 会立即抛出异常其他协程的结果全部丢失。# ❌ 危险写法一个失败全盘皆输resultsawaitasyncio.gather(*[fetch_url(u)foruinurls])# 如果第 3 个 URL 挂了前 2 个成功的结果也拿不到4.2 Wrapper 模式asyncdefsafe_fetch(url):try:resultawaitfetch_url(url)return(ok,result,url)# 成功返回成功标记exceptConnectionError:return(fail,None,url)# 失败返回失败标记resultsawaitasyncio.gather(*[safe_fetch(u)foruinurls])# 现在每个结果都带有状态标记不会丢失核心思想把异常变成返回值让业务逻辑自己决定如何处理失败。4.3TaskGroupvsgather wrapperPython 3.11 引入了asyncio.TaskGroup它在上层收集异常并抛出ExceptionGroup方案优点缺点gather wrapperAPI 简洁返回结果列表需要手动包装异常TaskGroup结构化并发更现代失败抛出ExceptionGroup处理复杂建议初学者从gather wrapper 开始理解后再尝试TaskGroup。五、完整代码 Python 异步编程实战并发抓取 URL 运行python async_fetch.py importasyncioimportrandomimporttimeasyncdeffetch_url(url:str)-str:模拟网络请求0.5~1.5s 延迟10% 失败率delayrandom.uniform(0.5,1.5)awaitasyncio.sleep(delay)ifrandom.random()0.1:raiseConnectionError(f请求超时:{url})returnf{url} 响应 (耗时{delay:.2f}s)asyncdefasync_fetch_all(urls:list[str])-tuple[list[str],list[str]]:并发抓取所有 URL单个失败不影响其他asyncdefsafe_fetch(url:str):try:resultawaitfetch_url(url)return(ok,result,url)exceptConnectionError:return(fail,None,url)resultsawaitasyncio.gather(*[safe_fetch(url)forurlinurls])success,failed[],[]forstatus,result,urlinresults:ifstatusok:success.append(result)else:failed.append(url)returnsuccess,failedif__name____main__:urls[fhttps://api.example.com/item/{i}foriinrange(1,11)]starttime.perf_counter()success,failedasyncio.run(async_fetch_all(urls))elapsedtime.perf_counter()-startprint(f✅ 成功:{len(success)}条)print(f❌ 失败:{len(failed)}条)print(f⏱ 耗时:{elapsed:.2f}s (同步需 ~10s))六、总结要点说明async/awaitPython 协程语法async def定义协程await挂起等待事件循环asyncio.run()启动负责调度协程asyncio.gather并发执行多个协程返回结果列表Wrapper 模式协程内部 try/except异常变返回值——单个失败不影响其他适用场景网络请求、文件 I/O、数据库查询等 I/O 密集型任务不适用场景CPU 密集型计算用multiprocessing代替Python 的异步编程上手不难难的是理解什么时候该用、怎么用得对。希望这篇文章能帮你少走弯路。本文基于 Python 3.10 语法编写完整代码可直接运行。