因为我平时给人改代码改得多发现一个很有意思的现象很多人写Python爬虫或者调接口代码逻辑没问题就是慢得要命。尤其是一次要处理几百个网络请求、几十个数据库查询的时候跑起来能从早等到晚。我最早也是这么过来的直到我把项目里那些同步I/O改成了async/await同一个任务从二十几分钟缩到了两分钟以内那会儿才真正意识到——非阻塞I/O调用不是锦上添花的技巧而是Python进阶路上绕不开的一道坎。这篇文章想聊的就是这件事async/await到底解决了什么问题底层怎么跑起来的以及你在真实项目里改造成异步时最容易踩的那些坑。适合学完Python基础语法、但一直没搞懂协程是怎么回事的读者。我会尽量把事件循环、协程调度、阻塞和非阻塞的差别用大白话讲清楚同时给出能直接抄走的改造思路和排查方法。1. 一次同步请求卡死全场的教训为什么要重新理解I/O先说个我今天还经常拿来当反面教材的例子。某开发者写过一个批量查询库存的系统逻辑很简单遍历一千个商品代码逐个调第三方接口拿到数据后写入数据库。第一版跑起来程序界面卡住不动进度条能一小时走不完。他一开始以为是接口限流后来抓包一看大部分时间都耗在等待响应上——每个请求发出去之后程序就死等在那里什么活儿都不干。这里的根本问题不在网络而在于同步I/O模型。1.1 I/O密集和CPU密集的差别直接决定你用不用异步写Python的人常听到“I/O密集型任务适合用异步”但很多人没细想过为什么。区分方法其实很简单如果你的程序大部分时间在等外部资源——等网络包回来、等磁盘读写完成、等数据库返回结果——那就是I/O密集型如果它大部分时间在占用CPU做运算、循环、处理数据那就是CPU密集型。同步代码处理I/O时线程会进入阻塞状态CPU时间片直接闲置。一次网络请求可能几十毫秒看起来不多但串行积累到一千个调用上就成了几十分钟。这个浪费是纯粹的等待不是计算。异步要做的就是在这些等待的缝隙里把CPU时间抢回来当一个请求在等网络响应时程序不闲着而是转头去发起下一个请求。这样一千个请求同时都在路上总耗时约等于最慢那一个而不是一千个加在一起。1.2 单线程并发听起来像玄学但是真的可行很多人第一次听到“单线程也能并发”时一脸怀疑觉得并发不是要多线程才行吗。这里要把概念拆开并行parallelism是多个任务真在同一时刻执行需要多核CPU或多线程而并发concurrency是多个任务在同一个时间段内交替推进只要切换速度够快宏观上看就是“同时在跑”。asyncio做的就是单线程内的事件循环并发。每个协程执行到await处就主动让出控制权事件循环把控制权交给下一个准备好的协程。网络请求的等待时间是并发的CPU计算依然串行但对我们遇到的大多数业务场景来说等待时间的压缩才是收益的大头。另外要说清楚一点很多人把Python的GIL和异步混为一谈。GIL限制的是同一时刻只能有一个线程执行Python字节码但I/O操作通常会释放GIL所以多线程在某些场景下也能并行等待。异步的额外优势是调度开销更小——线程切换要内核参与协程切换是用户态完成的成本低一个数量级。这也是为什么高并发I/O场景下异步方案往往比暴力起线程更稳。2. async/await的本质事件循环和协程的配合机制想用好async/await光会写async def和await不够得先理解幕后调度的那颗心脏——事件循环event loop。2.1 事件循环的工作方式像食堂打饭的窗口把事件循环想象成食堂里负责打饭的窗口阿姨。你点了菜发起网络请求阿姨记下订单告诉你“做好了叫你”注册回调、挂起协程然后立刻服务下一个同学。每个同学的菜好了阿姨就喊一嗓子那个同学过来端走恢复协程执行。整个食堂只有一个阿姨但所有同学不用干等着该干嘛干嘛。asyncio.run()是Python 3.7之后最推荐的程序入口它内部帮我们做了三件事创建新的事件循环、把传入的协程注册为入口任务、结束后关闭循环。我经常提醒初学者不要在async函数内部再嵌套一层asyncio.run()这会导致事件循环冲突报错非常经典。import asyncio async def main(): print(进入事件循环) await asyncio.sleep(1) print(协程恢复执行) asyncio.run(main())asyncio.sleep(1)这里特别值得琢磨。同步代码里的time.sleep(1)是真把线程挂起一秒钟期间什么都干不了await asyncio.sleep(1)则是告诉事件循环“我先歇着一秒钟后叫我”这一秒钟事件循环可以去跑别的协程。这个差异就是异步魔法的起点。2.2 可等待对象只有三类协程、任务、Future我在教学和代码审查里发现很多人的理解卡在“await后面到底能跟什么”上。记住这个结论就好await后面只能跟“可等待对象”awaitable而可等待对象就只有三种——协程对象coroutine、任务Task、Future。协程对象是async def函数调用后返回的东西。注意只调用不await它会给你一个警告协程从未被等待。很多人刚学的时候在这里翻车因为async def foo()和def foo()的调用习惯完全不同。async def fetch_data(): return 42 # 错误示范这样不会真的执行 coro fetch_data() # 正确做法要么await要么包成Task async def main(): result await fetch_data() # 或 task asyncio.create_task(fetch_data()) result await taskasyncio.create_task()是创建Task的正确姿势它会把协程包装成一个独立调度的任务立刻提交给事件循环。Task可以理解为“被事件循环盯上了的协程”。Future则是更底层的概念你可以把它当做一个“还没到手的值的占位符”通常你不会直接碰它但理解它有助于看懂第三方异步库的源码。2.3 await挂起点协程生命周期的关键转折await的真正含义是“在这里让出执行权”这也是事件循环能调度的基础。协程的执行过程可以看成开始执行 → 遇到await挂起 → 事件循环去做别的事 → 条件满足后被唤醒 → 从挂起点继续往下走。这个机制在Python里面是有状态支撑的。每个协程本质是一个生成器对象3.5之前的原始实现就是基于yield from协程的暂停和恢复就是生成器的send操作。await一个可等待对象时Python会在底层自动调用它的__await__方法把控制权交给事件循环。实际写代码的时候一个常见的误区是认为await能提速。await本身不做任何优化它只是给了别人“插队”的机会。真正提速的是你同时挂起了多个协程让它们在等待中重叠。一个人请求要等十秒你同时发了一百个请求总耗时还是十秒——这才是异步的收益逻辑。3. 一个真实改造案例把同步API调用批量改成异步光谈原理容易飘我拿一个具体场景完整走一遍改造流程。假设我们要抓取某个资源站的一千个页面每个页面解析后提取链接。同步代码写起来非常直白但跑起来就想砸电脑。3.1 改造前的代码模式看清阻塞点在哪import requests import time urls [fhttps://example.com/item/{i} for i in range(1000)] def fetch_sync(url): resp requests.get(url) return resp.text start time.perf_counter() results [fetch_sync(url) for url in urls] elapsed time.perf_counter() - start print(f同步耗时: {elapsed:.2f}s)这个版本的时间主要花在requests.get()的等待上。每个请求大约200毫秒一千个串行就是200秒上下。问题一眼就能看出来列表推导里的每次requests.get都让程序完全停住其他请求没法插队。3.2 改用aiohttp后的完整异步版本import asyncio import aiohttp import time urls [fhttps://example.com/item/{i} for i in range(1000)] CONCURRENCY 50 # 限制同时请求数量防止把对方打爆 async def fetch_with_semaphore(session, sem, url): async with sem: async with session.get(url) as resp: return await resp.text() async def main(): sem asyncio.Semaphore(CONCURRENCY) async with aiohttp.ClientSession() as session: tasks [fetch_with_semaphore(session, sem, url) for url in urls] results await asyncio.gather(*tasks) return results start time.perf_counter() results asyncio.run(main()) elapsed time.perf_counter() - start print(f异步耗时: {elapsed:.2f}s)改造后时间大约能压到5秒左右。有两个细节新手特别容易漏第一asyncio.gather(*tasks)接收的是多个位置参数所以列表要用星号展开第二信号量Semaphore(50)是必须的没有它一千个请求同时涌出去对方服务器大概率会拒绝服务你的IP还可能被临时封掉。3.3 为什么这里用gather而不是逐任务await有人会问不用gather改成循环里逐个await task行不行行但就变成串行了——每个await task都会等这个任务完成才继续下一个起不到并发效果。gather的语义是“一起等这批任务全部结束”是并发等待逐个await是“等完一个再等下一个”是串行等待。如果你需要“谁先完成就先处理谁”的场景比如爬虫里先拿到第一个响应就先解析可以考虑asyncio.as_completed()。它返回一个迭代器每次产出最早完成的任务适合做增量处理和进度刷新。二者各有适用场景我在下面做了个简单对比方法行为最佳使用场景asyncio.gather()等待所有任务完成返回结果列表批量请求后统一汇总asyncio.as_completed()按完成顺序逐个产出结果边下边解析、实时展示进度asyncio.wait()细粒度控制等待条件可设置超时需要手动管理任务集合的复杂场景4. 踩坑实录异步代码写起来顺跑起来全是意外异步改造不是把def换成async def、把requests换成aiohttp就万事大吉。我自己和身边人踩过的坑可以单独写一篇这里挑几个杀伤力最大的教训。4.1 异步代码里混入同步阻塞调用性能直接打回原形这是异步改造里最常见的反模式表面上是async def函数内部却调用了time.sleep()或者requests.get()这类同步阻塞库。事件循环被一个阻塞调用卡住整个循环里的所有协程全部停摆并发就剩个空壳。import asyncio import time import requests async def bad_fetch(url): # 错误time.sleep是同步阻塞会卡住整个事件循环 time.sleep(0.5) resp requests.get(url) return resp.text async def good_fetch(session, url): # 正确对应异步版本 await asyncio.sleep(0) async with session.get(url) as resp: return await resp.text()排查这类问题有个笨办法但很有效把并发数从1逐渐往上调观察耗时曲线。如果并发加到20后时间几乎没变多半是有同步阻塞调用混进去了。更省事的做法是写代码时定个规矩async def函数体内不允许出现time.sleep、requests.get、pymysql连接这类高频同步语句。4.2 创建了Task却忘了await代码里全是“Task was destroyed but it is pending”这个坑的报错信息相当有迷惑性——“Task was destroyed but it is pending”一看还以为是内存问题。其实原因是你用asyncio.create_task()创建了任务但事件循环结束前没来得及把它await完任务被强制销毁了。常见场景是在main()里create_task之后直接返回了没有gather或await那些Task。解决方式有两个要么在程序退出前统一await asyncio.gather(*tasks)要么确保create_task之后立刻被引用住并最终等待。千万不要把Task对象只是扔在列表里就结束主协程。这里还要说一个内存引用的问题。asyncio.create_task()会持有任务对象的强引用如果你创建大量Task后又不保留引用某些情况下任务可能被垃圾回收机制提前处理掉。稳妥做法是始终用一个列表或集合显式管理所有Task。4.3 回调风格和await风格混淆代码越写越乱很多人接触异步时先看的是旧教程里面大量使用add_done_callback这种方式处理结果。不是说它不行但和await混着用容易把代码结构搞得一团糟。await风格的核心价值就是让异步代码长得像同步代码顺序读下去就懂逻辑。我的建议很简单新代码一律用await风格。只有在对接不提供协程接口的第三方库时才需要用asyncio.get_event_loop().run_in_executor()把同步函数丢进线程池或者用loop.call_soon()这类底层API。前者博主级写法是import asyncio async def main(): loop asyncio.get_running_loop() # 把同步的requests.get丢到默认线程池执行避免阻塞事件循环 result await loop.run_in_executor(None, requests.get, https://example.com)这段代码解决了“第三方库没有异步版本”的尴尬场景。但要注意run_in_executor(None, func, *args)里的None表示用默认线程池如果任务量很大最好传入一个自定义的ThreadPoolExecutor来限制线程数。4.4 超时控制不做好gather永生等待异步代码并发能力上去了另一个问题就冒出来了某个请求如果卡死不返回await asyncio.gather(*tasks)会一直等下去整个程序像死锁一样挂着。同步代码里顶多一个请求超时抛异常异步代码里一个顽疾任务能让所有任务跟着陪葬。处理方式有几种最推荐的是给gather加return_exceptions参数或者用asyncio.timeout()Python 3.11包住整个等待过程import asyncio async def fetch(session, url): async with session.get(url) as resp: return await resp.text() async def main(): timeout asyncio.timeout(10) try: async with timeout: results await asyncio.gather(*(fetch(session, url) for url in urls)) except TimeoutError: print(整体超时干点别的逻辑)还有一点asyncio.wait_for(coro, timeout)也可以实现单个任务的超时控制但它的行为是取消这个任务。如果你用的是asyncio.timeout()配合asyncio.gather()取消行为更可控。工程上我习惯给每个请求单独设超时再给整体流程设一个兜底超时双保险。5. 并发上限、任务管理和其他工程化细节很多项目从“能跑”到“能上线”差的就是工程化的细节。异步代码尤其如此因为它的并发能力强了之后问题从“慢”变成了“不可控”。5.1 信号量限流别让你的程序变成攻击工具我见过不止一个新手的爬虫程序把别人网站打崩的。一千个请求同时冲过去对方服务器CPU直接跑满回来就封你的IP段。这不是技术问题是缺了基本限流意识。asyncio.Semaphore是异步版的信号量用法和threading.Semaphore类似区别在于使用async with sem而不是普通的with sem。每次进入协程先尝试获取信号量超过上限就等待其他请求完成后释放。async def bounded_fetch(session, sem, url): async with sem: # 这里面的代码最多同时执行CONCURRENCY份 async with session.get(url) as resp: return await resp.text()限流参数怎么定没有标准答案取决于目标服务的承载能力。比较稳妥的做法是从小往大试比如5、10、20。或者设置一个合理的延迟配合限流让请求均匀分布。别贪稳定才能跑得久。5.2 不要把CPU密集型计算放进协程里这是个特别容易混淆的点。异步适合I/O密集但如果你在一个协程里做了大量列表推导、循环、数据解析这些计算依然占用单线程的CPU时间。更麻烦的是这些计算不主动让出控制权事件循环里的其他协程就只能干等。处理CPU密集逻辑的标准姿势是交给asyncio.to_thread()Python 3.9或者进程池。前者把协程里的耗时计算挪到线程池执行适合中等负载后者适合真正的计算瓶颈但进程间通信会有额外开销。import asyncio def heavy_calc(data): # 模拟CPU密集 return sum(i * i for i in data) async def main(): data list(range(1000000)) # 不要直接在这里跑heavy_calc会卡住事件循环 result await asyncio.to_thread(heavy_calc, data) print(result)判断标准很简单如果一个操作耗时超过几十毫秒而且它不是I/O等待就要警惕它是否会拖垮整个事件循环。5.3 结构化和非结构化并发的选择从create_task到trio风格的启示asyncio里有句名言“如果你需要gather说明你的结构已经不对了。”这句话其实来自对结构化并发理念的倡导——与其手动管理一堆Task不如在代码结构上就把并发边界划分清楚。我自己的习惯是能用async with管理资源的地方绝不手动release能用TaskGroup管理任务的地方绝不裸用gather。Python 3.11起asyncio.TaskGroup已经是个很成熟的选择它在任务出错时的取消行为比gather更干净另一个任务抛异常时会自动取消组内其他任务避免僵尸任务残留。import asyncio async def main(): async with asyncio.TaskGroup() as tg: task1 tg.create_task(fetch(url1)) task2 tg.create_task(fetch(url2)) # 所有任务完成后继续执行有异常则统一处理这比手工写[asyncio.create_task(...) for i in range(n)]再gather要清晰得多。代码审查的时候看到老式的手工管理Task和回调混合代码我基本都会建议重构成TaskGroup版本。5.4 第三方库的异步适配情况一表看清做异步改造前先确认自己的核心依赖库有没有异步支持避免中途换方案。我整理了常见库的适配状态仅供参考库或工具同步版本异步支持情况备注HTTP客户端requestsaiohttp / httpx.AsyncClienthttpx支持同步异步双接口改造成本低数据库驱动pymysqlaiomysql / asyncmyasyncmy更快但生态不及aiomysqlPostgreSQLpsycopg2asyncpg性能很能打但接口差异不小MongoDBpymongomotor几乎无缝切换适合新手体验Redisredis-pyaioredis已并入redis-py异步接口新版redis-py自带async支持FastAPI/FlaskFlaskFastAPI / QuartFastAPI原生支持async端点一个替代思路如果你不想把所有代码全部改造成异步可以只在一个服务入口处使用异步比如用FastAPI写接口内部把同步逻辑丢给线程池。慢是慢一点但侵入性低适合老项目渐进式改造。6. 从同步到异步的心理转换与性能验证方法代码改造完成不等于任务完成。你需要验证异步真的带来了收益以及收益到底来自哪里。我实践下来有一套简单的验证方法。6.1 用粗暴实验判断瓶颈在哪里先跑一个基线版本同步代码处理一百个请求的耗时。再跑异步版本带信号量限流把并发数分别设为10、20、50记录耗时。如果并发从10调到50耗时基本不降那说明瓶颈不在并发等待而在于某个同步阻塞操作混在异步代码里。如果异步版本反而更慢先查是不是任务粒度太小了。协程并不是零成本的几千个微小任务在事件循环里来回切换开销可能超过收益。经验值是单个协程至少要处理一个需要几十毫秒以上的等待操作才值得异步化。6.2 事件循环里埋日志观察调度顺序调试异步代码和调试同步代码有个很大的不同错误堆栈不一定能帮你定位到任务的全貌。最实用的调试方法是在关键点加日志观察协程被唤醒的顺序。asyncio官方就提供了一个辅助工具asyncio.run()配合sleep(0)来强制切换或者使用PYTHONASYNCIODEBUG1环境变量启动程序它会启用事件循环的调试模式可检测到一些常见问题比如忘记await的协程和过长的回调。PYTHONASYNCIODEBUG1 python your_script.py这个模式下如果某个协程执行耗时超过100毫秒它会警告“Executing something took more than 100ms”这往往是混入同步阻塞或不合理的CPU计算的最直接信号。6.3 给协程写单元测试别把async当黑盒异步代码的测试和同步差异很大但没那么复杂。核心是unittest的IsolatedAsyncioTestCase或者pytest配合pytest-asyncio插件。测协程时直接在测试函数里await目标协程即可但要记得测试框架本身需要跑事件循环不是把测试函数写成async def就行。import asyncio import unittest class TestFetch(unittest.IsolatedAsyncioTestCase): async def test_fetch_returns_expected(self): result await some_async_function(test_url) self.assertEqual(result, expected)轻量级的脚本项目不想引测试框架也可以用asyncio.run()包一层来验证关键函数。不管哪种方式关键是别只测“不报错”要测并发行为本身。比如同时启动三十个任务看它们是不是真的交错执行了可以用asyncio.Event加asyncio.sleep(0)来制造可控的调度窗口。7. 再说几个补充的底层细节帮你写代码时心里有底这部分内容更多是理解性质的我平时写异步代码时用得上的底层模型整理出来分享给大家。7.1 async/await的关键字确实是语法糖但不是普通语法糖从实现上讲async def定义的是一个原生协程native coroutine它和生成器共享底层设施但行为不同——原生协程不能手动调用next()只能通过await驱动。Python的dis模块反汇编可以看出端倪async def函数的字节码带有ASYNC_GEN等指令标记编译器级别的支持远比字符串补丁式的实现高效。了解这层实现有个好处当你在网上搜资料时能看到很多老帖子讲“协程就是生成器”。那是Python 3.4时代的asyncio基于yield from的遗产。现代Python的协程已经是一条独立的实现路径不能简单用生成器的思维套用。7.2 Python版本和asyncio API的演进关系asyncio的API变化非常大网上很多教程还在教Python 3.6时代的写法比如asyncio.get_event_loop()配合loop.run_until_complete()。这些写法在新版本里不一定报错但已经有了更推荐的替代品。我建议读者直接按照你正在使用的Python版本查阅官方文档。Python 3.10之后很多老API标识为deprecated3.11引入了asyncio.timeout()和TaskGroup3.13进一步稳定了免GIL的多线程方案和部分异步接口。如果你读的教程经常让你用loop asyncio.get_event_loop()再手动管理循环多半可以换了。7.3 事件循环真的能“同时”处理多个socket吗这是初学者最喜欢问的问题之一。答案是事件循环靠操作系统提供的I/O多路复用机制来感知多个socket的状态。Linux上是epollmacOS上是kqueueWindows上是IOCP。asyncio的底层selector模块会把这些机制封装成统一的接口事件循环只需要问“哪些socket可读/可写了”然后挨个处理。所以你看到的一千个并发连接实际上是事件循环在高频轮询“谁就绪了”。这也是为什么await后面接的必须是异步I/O普通的requests.get不会向事件循环注册任何socket事件事件循环根本不知道该等谁。只有aiohttp这种库内部用loop.create_connection()注册了回调才能实现真正的非阻塞等待。8. 踩了无数坑之后我会怎么开始一个新项目最后聊点个人习惯。现在我接到一个新的I/O密集型项目第一步不是写代码而是先想清楚这个项目的瓶颈在哪里是等待多还是计算多需要依赖什么第三方库它们有没有异步支持有没有并发安全的问题需要提前考虑。想好了之后我会先写一个最小可运行的异步原型比如只是同时发十个请求确认链路通了再慢慢加并发和业务逻辑。说白了异步不是银弹它的价值在于把骑虎难下的等待时间拿回来。适合用异步的场景收益非常明显不适合的硬上异步反而会把简单问题复杂化。如果你正在学这一块我建议你从一个小爬虫或者接口轮询场景开始练手先把asyncio.run、create_task、Semaphore、gather这四样跑熟再去看事件循环的源码也不迟。代码这东西读一百篇教程不如动手跑通一个异步Demo跑通之后你再看任何开源项目的异步代码都会觉得亲切得多。
