如果你写过需要同时请求几十个外部接口的脚本大概率遇到过这样的场景程序明明只是等对方响应CPU却空转了大半时间一个请求慢后面的全被拖死。Python里解决这种“白等”问题的正统方案就是async和await这套非阻塞I/O调用机制。这篇想和你聊聊我实际用下来的理解为什么非阻塞I/O值得关心、async/await背后到底发生了什么、以及怎么把一套可用的异步代码真正落地到脚本和项目里。适合已经会写Python、被阻塞I/O卡过脖子、想系统理清协程和事件循环的进阶学习者看完整理出属于自己的那套套路。1. 先想明白阻塞到底浪费了什么1.1 一个同步请求的时间都去哪了很多人第一次感受到“阻塞”的痛是从爬虫开始的。写一个最简单的循环抓链接用同步方式请求几十个页面总耗时长到像程序死了一样。问题出在哪我拆过一个普通HTTP请求的耗时构成DNS解析、TCP握手、TLS握手、发送请求、等待服务器处理、接收响应体。真正的本地计算少得可怜绝大多数时间都在“等网络”。同步代码遇到等网络时线程是睡着的。它不占CPU但它把执行流卡死了。你如果用requests在for循环里发请求等于一个人站在柜台前等着上菜其他桌的菜全没人管。生活里这样排单会饿死人代码里这样写会浪费掉大量时间窗口。我用一个很粗的比喻记住这件事同步阻塞是人等事非阻塞I/O是事等人。后端把任务挂起来网络响应什么时候回来什么时候再捡起来继续算。中间的空档调度器可以拿去做别的请求这就是并发的来源。1.2 多线程不是银弹线程切换有成本有人会说麻烦事交给多线程不就行了主线程发一堆线程每个线程各自同步等待。理论上对但Python的场景里有两个现实问题。第一个是全局解释器锁GIL。在CPython里同一时刻只有一个线程在解释器里跑Python字节码。当然I/O等待时会释放GIL所以多线程做网络请求确实能提速这点不能否认。Python的threading并不是一无是处。但第二个问题更隐蔽线程有调度成本。每开一个线程操作系统要分配栈空间、维护内核线程对象上下文切换时保存和恢复寄存器、切换栈。线程量上去以后调度器本身就在消耗CPU。写一个爬虫开几百个线程试试你会发现前面提的速一部分又被切换开销吃回去了。线程太多还会撞上文件描述符、内存等系统资源上限。事件循环的思路不一样它只在一个线程里跑调度不依赖操作系统去切换线程而是让任务主动让出执行权。这个主动让出的机制就是await。成本极低一个状态机切换而已。所以同样一份代码用异步写出来往往能用更少的资源撑住更大的并发。2. 把async和await这两张牌拆开看2.1 async def定义的其实不是“异步函数”这是新手最容易绕晕的地方。你在函数前面加了async它并不变成一个立刻能跑的函数。直接调用它返回的是一个协程对象。async def say_hello(): return hello print(say_hello()) # coroutine object say_hello at 0x...看到没有函数体根本没有执行。协程对象可以理解成一个暂停的计算流程它定义了“之后要做什么”但必须有人去驱动它。驱动方式有三种写一个await去等它把它交给asyncio.create_task变成任务或者直接用asyncio.run去跑。如果你在普通的同步代码里拿到一个协程对象又没去await、也没交给调度器Python会告警说这个协程从未被等待。原因就是“流程已经搭好但没人推它往前走”。我自己的理解方法是async def是生产协程的工厂协程对象是一张需要被消费的待办清单。只有事件循环真正执行到它、遇到await挂起清单上的步骤才逐步推进。2.2 await到底在等什么await后面跟着的必须是可等待对象。常见的有三类协程、Future、Task。协程另一个async函数返回的协程对象。Future一个低层对象代表“未来某个时刻会有的结果”。可以把它想成一个快递单号查物流的状态物流没到就等到了就取件。Task它是Future的子类专门用来包装协程把“暂停的流程”提交给事件循环调度。await一个协程是把它嵌入当前流程必须等它跑完才能继续。await一个Task同样会等它完成但Task背后的事件循环可能还在跑别的任务。这两者最大的差别在于“并发”你在一个async函数里写出await another()这是顺序执行如果你先把多个协程create_task,再await这些任务它们就在事件循环里交错推进。async def main(): # 顺序执行索性等了a再等b await task_a() await task_b() # 并发执行两个任务都提交给事件循环 task1 asyncio.create_task(task_a()) task2 asyncio.create_task(task_b()) await task1 await task22.3 事件循环真正的那个幕后调度员async/await本身只是语法层面的“挂起约定”真正干活的是事件循环。事件循环运行在单线程里它维护着一个就绪队列不停问谁准备好继续跑了你可以在一个Python进程里创建多个事件循环但实际上绝大多数项目只需要一个。import asyncio async def demo(): print(开始) await asyncio.sleep(1) print(结束) asyncio.run(demo())asyncio.run干的事情可以拆成三步创建一个新的事件循环在循环里运行传入的协程直到完成后关闭清理所有未完成的任务。和旧时代手工写loop.run_until_complete()再手动close()相比run把入口统一了。记住一个重点事件循环所在的线程不能被阻塞。你如果在协程里调了time.sleep(3)事件循环整个被卡住3秒所有其他任务都得等着。这也是“非阻塞”的另一面不是你的业务代码一定不阻塞而是别在调度线程里做阻塞事。3. 动手写一个非阻塞I/O调用同步改异步全程实录3.1 先做一条同步基线我们假设要抓取5个接口的数据接口平均响应0.2秒。同步版本长这样import time import requests URLS [ https://httpbin.org/get?i0, https://httpbin.org/get?i1, https://httpbin.org/get?i2, https://httpbin.org/get?i3, https://httpbin.org/get?i4, ] def fetch_sync(): results [] for url in URLS: resp requests.get(url, timeout3) results.append(resp.json()[args]) return results if __name__ __main__: start time.perf_counter() data fetch_sync() print(耗时:, round(time.perf_counter() - start, 2), 秒)我本机跑大概是1秒多。这里的核心问题不是慢而是“白等”一个请求在网络上等200毫秒CPU一点事没干后面的请求只能排队。3.2 异步改写从requests换到aiohttprequests是同步库在协程里直接调用会卡住整个事件循环必须换成异步的HTTP客户端。我用aiohttp比较多因为它成熟、API稳定。import asyncio import aiohttp import time URLS [ https://httpbin.org/get?i0, https://httpbin.org/get?i1, https://httpbin.org/get?i2, https://httpbin.org/get?i3, https://httpbin.org/get?i4, ] async def fetch_json(session, url): async with session.get(url, timeout3) as resp: return await resp.json() async def fetch_async(): async with aiohttp.ClientSession() as session: tasks [asyncio.create_task(fetch_json(session, url)) for url in URLS] results await asyncio.gather(*tasks) return [item[args] for item in results] if __name__ __main__: start time.perf_counter() data asyncio.run(fetch_async()) print(耗时:, round(time.perf_counter() - start, 2), 秒)这段代码里我用了aiohttp.ClientSession创建连接池每个URL都创建一个Task最后用asyncio.gather等所有任务完成。gather的厉害之处在于传入多少个任务它就返回多少个结果而且返回顺序和传入顺序一致这给结果归组省了很多事。跑完大概0.3秒左右同步版本1秒多3到4倍的提升。而且随着请求数量增加收益会更明显。30个请求同时打出去同步版本可能要6秒异步版本可能只要1秒多。3.3 生产环境还不够超时、并发限制、异常隔离上面的代码能跑但直接扔生产会撞出问题。第一个就是并发量。如果你的URL有上千个一次性把所有Task都创建出来很可能瞬间打爆目标服务的连接数甚至被防火墙拦截。解决办法是引入信号量限流。semaphore asyncio.Semaphore(20) async def fetch_with_limit(session, url): async with semaphore: return await fetch_json(session, url)这个信号量等于“最多20个并发窗口”多余的协程会排队等着。触发信号量的协程会挂起事件循环继续调度其他协程所以限流不会浪费等待时间。第二个是超时。我上面的代码里给session.get加了timeout3这是aiohttp的连接和读超时。但更稳妥的做法是用asyncio.wait_for包裹整个请求协程给它一个硬性时间上限。try: result await asyncio.wait_for(fetch_json(session, url), timeout5) except asyncio.TimeoutError: result None第三个是异常隔离。一个请求挂了不该让整个gather炸掉。asyncio.gather有return_exceptions参数设为True后异常会被当成返回值返回而不是抛出。results await asyncio.gather(*tasks, return_exceptionsTrue)这个参数帮过我好几次批量抓数据时偶尔一个接口5xx其他请求的数据照常收回来我只需要在结果里筛掉异常类型即可。注意异步代码里的异常不会自动打印堆栈。协程被挂起、被调度、再恢复栈信息和普通同步调用不太一样。排查问题时先别急着猜业务逻辑先看异常是在哪个任务里抛的。4. 库选型与同步代码桥接不是所有代码都得是异步的4.1 异步HTTP客户端怎么选只要涉及外部网络请求就必须用支持异步的客户端库。我常用的有aiohttp和httpx。两者看起来差不多但应用取向有差异。特性aiohttphttpx核心能力服务端/客户端双端Web框架也能搭客户端为主接口设计贴近requests连接池管理内置支持Cookie、代理配置内置支持HTTP/2同步/异步双模主要异步同步、异步都支持上手难度稍高需要理解session上下文很低代码像requests维护活跃度很稳定新特性更新更快我的建议是项目里已经有aiohttp依赖或者你要写高并发爬虫直接选aiohttp如果你只是想把原有requests代码渐进式迁移到异步或者追求最简单的心智模型优先httpx。AsyncClient的用法和requests真的很像import httpx async def fetch_with_httpx(): async with httpx.AsyncClient(timeout10) as client: resp await client.get(https://httpbin.org/get) return resp.json()还有一个常见场景是异步文件读写。普通open()是同步阻塞的协程里直接读一个大文件也会卡事件循环。异步文件库aiofiles可以解决这个问题但文件I/O通常不是网络场景的最大瓶颈如果你的并发需求不大用asyncio.to_thread跑文件读取也足够了。4.2 事件循环里别碰阻塞调用桥接同步库有正经姿势新手最容易踩的坑在async函数里直接调用同步阻塞库。最常见的“自杀式写法”是下面这样async def bad_example(): # requests.get是同步阻塞会卡住整个事件循环 resp requests.get(https://httpbin.org/get, timeout3) return resp.json()这段代码跑起来其他所有并发任务都会卡住等于用异步的壳包了一个同步的实现。为什么慢因为事件循环是单线程的你用同步库等于把一个耗时操作塞进了调度线程整个循环停摆。正确的桥接姿势是用asyncio.to_thread把同步函数丢到线程池里跑import requests async def good_example(): resp await asyncio.to_thread(requests.get, https://httpbin.org/get, timeout3) return resp.json()asyncio.to_thread是Python 3.9引入的便捷封装底层走run_in_executor默认线程池。如果你的同步函数本身是CPU密集型的用线程池反而受GIL限制应该考虑run_in_executor配ProcessPoolExecutor但那个已经是“并行”的范畴不是本篇的核心。我想强调一个边界不要为了异步而异步。一个简单脚本、只有几个请求、逻辑不复杂用同步代码可能更清晰。异步的价值在“大量I/O等待可并发”时才显现。你如果把一个本来10毫秒跑完的函数强行改成协程不仅没有收益还引入调度复杂度。先画清楚瓶颈在哪再决定要不要动刀。5. 常见问题与排查技巧实录5.1 你迟早会撞见的几个报错我把自己和身边几位朋友踩过的坑整理成一张速查表按常见程度排序。报错或现象常见原因排查/解决方案RuntimeError: asyncio.run() cannot be called from a running event loop在已运行的循环里再调用asyncio.run常见于Jupyter或嵌入式脚本在事件循环内部用asyncio.create_task提交别再runTask was destroyed but it is pending协程被挂起后还没跑完事件循环就关了确保所有任务都被await或取消退出前先gatherasyncio.TimeoutError请求超时调整timeout或重试区分连接超时和读超时aiohttp.ClientConnectorError连接被拒绝、DNS失败检查域名、端口、网络连通性加代理或重试机制RuntimeError: Event loop is closed多次使用同一个关闭的循环Python 3.10以后asyncio.run会重建循环避免手动管理loop对象程序“卡死”无输出事件循环被某处阻塞调用拖住全局搜time.sleep、requests.get、open()等同步调用替换为异步或to_thread这里重点说task被销毁的情况。如果你在main里create_task后没有等它跑完事件循环关闭时会强制销毁任务报出warning。我看过很多初学代码在asyncio.run里只await了第一个任务后面几个任务变成“孤儿”。正确做法是永远收集所有需要跑的任务最后一起await。async def main(): tasks [ asyncio.create_task(worker(i)) for i in range(10) ] await asyncio.gather(*tasks)5.2 调试异步代码的几条野路子异步代码的问题排查比同步代码难受因为调用链不连续。我摸索出的第一招是开asyncio调试模式。在运行脚本前加上环境变量PYTHONASYNCIODEBUG1 python your_script.py这会帮你检测到不少问题比如“协程从未被await”、“慢回调”等。还可以在创建事件循环时手动设置import asyncio async def main(): loop asyncio.get_running_loop() loop.slow_callback_duration 0.05 # 超过50ms的回调会告警 # ...这个slow_callback_duration很有用。什么概念事件循环要求每个回调执行得很快一旦有某个回调执行超过阈值它就意味着阻塞了。把阈值设小一点你能在日志里看到哪些回调拖了后腿。第二招是给每个任务加上下文日志。协程并发时print输出顺序会混在一起你很难看清某个任务跑到哪了。我习惯在任务入口和出口各留一条日志任务标识用URL或业务ID这样眼里看到的就不是一串乱码。async def fetch_with_log(session, url): print(f[{url}] 开始) try: data await fetch_json(session, url) print(f[{url}] 完成) return data except Exception as e: print(f[{url}] 失败: {e}) raise第三招是注意取消。超时取消后协程内部如果有finally它会跑一遍。如果你的协程里持有文件句柄、数据库连接记得在finally里释放。异步的with语句天然支持清理所以尽量把资源获取放在async with里。第四招别把所有业务逻辑都堆在一个大协程里。协程是串联的如果一个大协程内部一直在await各种子协程那它本质上还是顺序的。你要并发就得在尽可能高的层级切分出独立任务。我通常的做法是写一个小协程负责“单条流水线”主协程负责“创建任务并聚合结果”。5.3 性能调优从“能跑”到“跑得顺”异步程序性能调优我一般按四个维度查。第一看并发上限。信号量太小并发不够时间提升不明显信号量太大目标服务报错资源浪费。具体值要根据目标服务的吞吐能力和响应时间压测得出。我习惯从20开始逐步往上试探。第二看连接池配置。aiohttp的ClientSession默认连接数限制是100如果你要发上千请求需要调TCPConnector的参数connector aiohttp.TCPConnector(limit50, limit_per_host10) async with aiohttp.ClientSession(connectorconnector) as session: ...limit是总连接数limit_per_host是单个域名的连接上限。这两个参数和信号量是两套机制不要混在一起调。信号量控制“同时跑的协程数”连接池控制“连通的对端连接数”两者配合效果好。第三看重复I/O。很多接口可以一次批量查询就别拆成几十个小请求。网络请求里握手、序列化、协议开销是固定的请求数越多这些固定开销占比越大。这不是异步能解决的要从接口设计层面优化。第四看结果处理。gather返回后如果你还要做很多CPU计算可以考虑把计算部分丢给线程池。异步解决了I/O等待不等于所有慢操作都变快。CPU密集仍然是CPU密集该用进程池还是用进程池。结尾想说的话用async/await写非阻塞I/O它不是某种新语言特性那么简单更多是一种心智模型的切换。我第一次从同步改异步时总想“这个await放在这还是放那”后来才明白await不是等待完成而是把控制权交还给事件循环让其他任务趁机推进。想通这一点很多代码结构就自然顺畅了。个人实操中还有个小技巧值得分享遇到一个卡顿的脚本先别急着上异步。先用同步版本写通业务逻辑确认数据流没问题再逐块改成协程。你保留一个可运行的同步基线后面排查异步问题时能随时对照这个底牌特别重要。异步代码本身的难点不在API而在并发状态的推理。公事公办把人等事的思维换成事等人的模型剩下的就是多写几次的事。