最近群里聊Python出现频率最高的词就是“异步”。不少人刷到过“一行代码让你的爬虫提速十倍”之类的标题自己跟着写了半天要么没效果要么一跑就报错甚至连async和await到底干了什么都没搞明白。说实话这个问题我踩了两年坑才理顺。异步确实能带来肉眼可见的提升但前提是理解它到底改写了哪部分逻辑。这篇文章就以asyncio库为主线从最底层的“为什么慢”讲起到事件循环的运行机制再带一个完整的改造案例最后把新手最容易翻车的几个现场逐一拆开。适合刚接触异步、或者写过一点async/await但总感觉飘着没落地的读者。我不打算从“什么是协程”这个概念讲起先讲一个更根本的问题你的程序到底慢在哪。1. 协程要建立的第一个直觉慢不在计算在等待很多人把“异步”当成玄学是因为根本没搞明白同步代码的瓶颈在哪里。计算机执行代码的速度极快哪怕是低配机器一秒跑几百万次普通运算也没压力但I/O操作不一样——读写文件、访问数据库、发起HTTP请求每一次都在和数据源打交道而数据源可能是远在几千公里外的服务器也可能是一块物理硬盘。试着感受一下这个场景你写一个脚本从某网站批量获取数据串行跑200个请求平均每个请求要400毫秒才能响应。这200个请求跑完就是80秒。如果观察这段时间CPU在干什么——它基本是空闲的。因为每个请求发出之后程序就傻傻地挂在那儿等服务器回包这400毫秒里CPU几乎什么都没做就是“等”。这是个关键认知同步程序慢不是因为它做的事情多而是因为它把时间花在了等待上。打个比方。你去小吃店点了一份炒粉、一份汤、一份甜品。如果采用“点一份、站那儿等它做好、吃完、再点下一份”的方式你花的时间等于三份食物的制作时间之和。更聪明的做法是把三份一起下单然后坐在位置上等哪个好了服务员喊你一声你就去端。后面这种做法就是异步。协程要解决的正是这个“傻等”的问题。它并不让单个请求变快它让程序在等待一个请求返回的时候不干站着而是顺手去发下一个请求、做下一件能做的事。所以在动手写asyncio之前先对着自己的项目问一句我的任务是不是I/O密集型的网络请求、文件读写、数据库查询、消息队列消费都算。如果你的任务是CPU密集型的比如图片处理、大规模数值计算、文本解析那异步带来的收益会非常有限因为这类任务没有“等待间隙”可以利用CPU一直在满负荷工作协程切来切去反而增加开销。这也是很多初学者把异步用在错误场景后觉得“什么破玩意儿”的根源。异步编程的核心心法就是一句话把等待的时间变成干活的时间。后面所有代码、所有概念都是围绕这句话展开的。2. 事件循环是怎么把“等待时间”拿回来复用的理解了“慢在等待”之后就该看看asyncio拿回等待时间的具体手段。这里绕不开三个关键词协程、任务、事件循环。如果你把它们的关系理清了后面写代码基本就是顺手的事。2.1 协程只是一个函数不是特效先看最简单的代码import asyncio async def say_hello(): print(hello) await asyncio.sleep(1) print(world) asyncio.run(say_hello())async def定义的不是一个普通函数而是一个协程函数。调用它会返回一个协程对象但函数体内部的代码不会立刻执行。你如果直接跑say_hello()Python解释器只会给你一个警告coroutine say_hello was never awaited。这就好比食堂窗口做好了饭菜但窗口迟迟没有轮到你也没人叫你的号。真正让协程跑起来的是asyncio.run()它做了三件事创建事件循环、把协程喂进去运行、结束后关闭循环。这是Python 3.7以后官方推荐的标准入口日常写应用用一个asyncio.run(main())就够了完全不需要手动管理事件循环。2.2 事件循环就像一个叫号排队的调度台事件循环是整个asyncio的调度核心。可以把它想象成一个餐厅的排号调度器桌上放着很多等位牌协程任务调度器不断循环检视哪个号的菜还没好就让出CPU哪个号的菜好了就叫号继续执行。await关键字的意思是“在这里挂起”。遇到await asyncio.sleep(1)时协程把控制权交还给事件循环事件循环立刻去调度其他就绪的协程。等一秒钟到了sleep完成事件循环回来唤醒这个协程让它接着往下执行。你可能会问多线程不也能做类似的事情吗确实能但线程的切换耗的是操作系统的调度开销而且线程多了以后还要面对锁、竞态条件、线程安全问题。协程的切换发生在一个线程内部完全没有系统调用的开销切换成本极低这才使得“你发几万个请求不会把机器跑挂”成为可能。2.3 任务Task是协程的“带号码牌”版本协程对象本身不能被事件循环直接调度你需要把它包装成Task。一个很常见的写法是async def fetch_data(num): await asyncio.sleep(num) return num async def main(): task1 asyncio.create_task(fetch_data(1)) task2 asyncio.create_task(fetch_data(2)) result1 await task1 result2 await task2 print(result1, result2) asyncio.run(main())create_task会立刻把协程注册到事件循环里让它在后台排队。到了await task1程序才停下来等这个任务的结果。如果多个任务都已经创建那么事件循环会穿插执行它们而不是按顺序老老实实地排队等。这里有一个新手很容易误会的点create_task之后就必须await否则任务会跑完就消失了甚至可能在事件循环关闭时收到Task was destroyed but it is pending的告警。任务不是“创建完就没人管了”你至少要用一个列表把它们都装着最后gather或逐个await程序才会稳定。asyncio.gather是另一套聚合等待方案results await asyncio.gather(task1, task2)它接收多个可等待对象返回对应顺序的结果列表。比起手写两个awaitgather在处理“同时等一大堆任务”时简洁得多后面实战里会大量用到。3. 一个完整改造案例同步批量请求到异步并发概念讲了半天不如直接上手改个项目。我这里用“批量请求公开接口获取数据”的场景来做示范假设你有200个URL需要把每个URL的响应内容长度统计出来。先写同步版本再改成异步版本最后对比效果。3.1 同步版本跑起来你会看到CPU在打瞌睡import time import requests urls [fhttps://httpbin.org/bytes/{i} for i in range(100, 300)] start time.perf_counter() total 0 for url in urls: resp requests.get(url, timeout10) total len(resp.content) elapsed time.perf_counter() - start print(f同步版耗时: {elapsed:.2f}s, 总字节数: {total})requests是同步库requests.get发出请求后当前线程被阻塞直到响应返回才继续。如果每个请求平均响应300毫秒200个请求串行跑就是60秒。这60秒里CPU绝大多数时间是空闲的。3.2 异步版本用aiohttp替换requestsimport asyncio import time import aiohttp urls [fhttps://httpbin.org/bytes/{i} for i in range(100, 300)] async def fetch_one(session, url): async with session.get(url, timeout10) as resp: data await resp.read() return len(data) async def main(): async with aiohttp.ClientSession() as session: tasks [fetch_one(session, url) for url in urls] lengths await asyncio.gather(*tasks) print(f异步版总字节数: {sum(lengths)}) start time.perf_counter() asyncio.run(main()) elapsed time.perf_counter() - start print(f异步版耗时: {elapsed:.2f}s)如果你的网络条件正常这个异步版本通常能把跑200个请求的时间压到同步版的五分之一甚至十分之一。我第一次跑的时候同步版用了七十多秒异步版只用了八秒那会儿才真切体会到“等待时间换成干活时间”是什么感受。这里有个细节值得说明aiohttp.ClientSession是异步上下文管理器session.get返回的响应对象也是。不要把aiohttp和requests混着用。aiohttp是Python生态里最常用的HTTP异步客户端如果你做的是爬虫类项目它基本是标配。为什么不用requests因为requests是纯同步库它的所有调用都会阻塞事件循环放进异步代码里就是灾难。后面第4章会专门讲这个坑。3.3 瓶颈在哪连接复用与并发数异步代码跑得快不代表没有优化空间。aiohttp.ClientSession底层维护了一个连接池同一个session会自动复用TCP连接避免每次请求都重新握手。如果你每次请求都新建一个ClientSession性能会大打折扣这点和requests.Session的道理一样。另一个隐藏瓶颈是并发数量。上面的代码把所有任务一股脑丢给gather200个并发请求同时发出。目标服务器如果扛不住可能会拒绝服务。正经做批量请求的时候一定要加信号量控制并发我在第5章会给出限流版本。跑通这段代码之后你对asyncio的“非阻塞”应该会有直观感受请求发出后事件循环没有等待而是立刻去执行其他待处理的协程直到某个响应回来才接着处理。4. 入门期的四个典型翻车现场症状、根因与纠正现在说说踩坑。下面这些错误我基本都犯过而且每次都是在线上环境里被“突然出现的诡异行为”教训了一顿才彻底记住。它们有一个共同点语法上完全合法程序不报错但结果完全不对——这类问题最恶心。4.1 现场一在async函数里用了time.sleepimport asyncio import time async def worker(): print(start) time.sleep(3) print(end) async def main(): await asyncio.gather(worker(), worker())直觉上time.sleep(3)就是“睡3秒”但这两个worker()实际执行耗时不是3秒而是6秒。原因在于time.sleep是同步阻塞函数它会让整个线程停下来事件循环没有机会去调度另一个worker和原来的同步程序没有任何区别。正确做法是把time.sleep换成asyncio.sleep。这是异步代码里最容易犯的第一大错因为同步写法思考得太顺手了。4.2 现场二在async代码里调用了requests这个坑一半是上面那个的进阶版。requests的get会阻塞线程所以在异步程序里只要有一个请求用requests.get整个事件循环就会卡住后面所有排队的协程全部陪着等待。典型的症状是你明明用了asyncio.gather跑了100个任务总耗时却和同步版本一模一样甚至更慢。如果某个同步库确实绕不开比如只有它支持某个老接口有两个处理办法import asyncio import requests async def fetch_with_requests(url): loop asyncio.get_running_loop() return await loop.run_in_executor(None, requests.get, url)run_in_executor会把同步函数丢给线程池执行不阻塞事件循环。Python 3.9之后还有一个更简洁的写法async def fetch_with_requests(url): return await asyncio.to_thread(requests.get, url)不过要记住这属于“退而求其次”跑的是普通线程并发量上不去只是让事件循环不至于卡死。4.3 现场三Task创建后忘记保存引用import asyncio async def job(): await asyncio.sleep(5) print(done) async def main(): for _ in range(10): asyncio.create_task(job()) await asyncio.sleep(6) asyncio.run(main())这段代码里每个job()都会被创建但没有任何变量引用它。Python的垃圾回收机制会认为这些Task“没人要”在它们运行完之前就把它们回收掉。轻则任务根本没跑重则事件循环关闭时打印一堆Task was destroyed but it is pending。正确做法是用一个列表把Task保存起来async def main(): tasks [asyncio.create_task(job()) for _ in range(10)] await asyncio.gather(*tasks)4.4 现场四在多个线程里重复调用asyncio.runasyncio.run每次都会新建一个事件循环同一个事件循环对象绑定到创建它的线程。如果程序通过ThreadPoolExecutor在多个线程里各跑一个asyncio.run(main())表面看可行但Python 3.10会抛出RuntimeError: asyncio.run() cannot be called from a running event loop。还可能出现“event loop is closed”之类的玄学异常。如果你真的需要在另一个线程里跑异步代码正确姿势是预先开好一个事件循环并运行import asyncio import threading loop asyncio.new_event_loop() def run_loop(): asyncio.set_event_loop(loop) loop.run_forever() threading.Thread(targetrun_loop, daemonTrue).start()但实际业务里这种场景很少。绝大多数应用一个主事件循环就够了如果你发现自己要“多线程asyncio”先停下来想想架构是不是绕远了。这四个翻车现场覆盖了90%的入门事故。如果你照着写代码时遇到了诡异行为先按这个清单自查一遍是不是在异步里用了同步阻塞调用是不是Task没保存是不是循环生命周期没管理好5. 限流、超时、取消让你的异步程序收得住别急着庆祝。新手把同步改异步之后第一个惊喜是“快了”第二个惊吓是“把服务器打挂了”或者“有大量任务卡住不返回”。一个能上生产的异步程序必须能控制并发、管住超时、处理取消。5.1 用Semaphore限制并发数asyncio.Semaphore相当于流量阀门。它内部维护一个计数器拿到信号量才能继续执行执行完释放一个名额。典型用法import asyncio import aiohttp URLS [fhttps://httpbin.org/bytes/{i} for i in range(100, 300)] SEMAPHORE asyncio.Semaphore(10) # 最多10个并发 async def fetch_one(session, url): async with SEMAPHORE: try: async with session.get(url, timeout10) as resp: data await resp.read() return len(data) except Exception as exc: print(f请求失败 {url}: {exc}) return 0 async def main(): async with aiohttp.ClientSession() as session: tasks [fetch_one(session, url) for url in URLS] result await asyncio.gather(*tasks) print(f总字节数: {sum(result)}) asyncio.run(main())设置并发上限的意义不只是“对服务器友好”也是给自己留后路。并发过高会导致大量请求超时、连接被重置最终整体吞吐量反而下降。根据目标网站的承受能力来定信号量常见做法是先压到5、10、20试几轮观察延迟和失败率再调整。5.2 用wait_for给每个任务加超时有时候某个请求就是死活不返回你没有超时控制整个gather就会一直挂下去。asyncio.wait_for可以给协程包一层超时async def fetch_one_with_timeout(): return await asyncio.wait_for(fetch_one(...), timeout15)超时后会抛出asyncio.TimeoutError别忘了在gather外层捕捉。也可以用asyncio.timeout()上下文管理器Python 3.11语义更清晰try: async with asyncio.timeout(15): data await fetch_one(...) except TimeoutError: print(该请求超时)5.3 取消任务不是你想取消就能取消干净task.cancel()会向任务内部注入CancelledError。协程收到这个异常后可以选择响应、清理资源然后退出也可以选择忽略但不推荐。如果你在协程里打开了一个文件或者持有一个连接正确做法是用try/finally保证资源释放async def work(): try: async with some_connection() as conn: await conn.query(...) except asyncio.CancelledError: # 取消时先释放资源再退出 raise注意CancelledError在Python 3.8之后继承自BaseException不再继承Exception所以如果你用except Exception去捕获是抓不到取消信号的这也是个常踩的坑。5.4 gather与异常的一揽子处理gather有一个常用参数return_exceptionsTrue。设置后协程里抛出的异常不会中断整个gather而是作为结果值返回results await asyncio.gather(*tasks, return_exceptionsTrue)处理批量任务时这是个救命选项。否则一个任务挂了所有任务跟着“陪葬”你都不知道谁成功谁失败。拿到结果列表后再用isinstance(result, Exception)做判断即可。Python 3.11引入的TaskGroup则是更高级的替代品它自动管理任务生命周期取消一组任务时更干净。如果你用的是新版本Python建议了解一下async def main(): async with asyncio.TaskGroup() as tg: for url in URLS: tg.create_task(fetch_one(session, url))6. 什么时候别用async兼谈我的推荐路线最后必须泼一盆冷水异步不是银弹不是所有程序都适合改成异步也不是所有库都支持异步。如果项目只是几十个请求的简单脚本同步完全够用改成异步纯属增加阅读负担。如果任务是CPU密集型的比如图片识别、视频转码、大数据集运算用asyncio帮倒忙你应该考虑的是multiprocessing靠多进程把CPU核心用满。如果代码是几十万行的大型业务系统里面大量使用同步库你硬要“全局异步化”会遇到一个接一个的兼容性问题改到怀疑人生。异步是主动设计出来的架构不是后期随便“加上去”的补丁。现阶段比较务实的路线是先熟练掌握async/await、asyncio.run、gather、Semaphore、wait_for、to_thread这几个核心工具。日常写脚本时遇到I/O密集型的批量任务优先考虑aiohttp这类支持异步的库。把协程理解成“能挂起、能恢复、由事件循环调度的函数”出了问题先怀疑阻塞调用是排查异步Bug最有效的思路。我在实际项目里用得最多的一招是把异步代码封装成一个独立模块暴露同步接口给调用方。内部用asyncio.run兜底外部完全感知不到异步的存在。这样既能享受异步的提速又不至于让整个项目被协程绑架。等到后续团队需要上量了再把这些模块逐步迁移到长驻事件循环里。异步学习最忌讳的就是套模板背概念。多跑几个例子多看看任务对象在事件循环里的调度顺序你很快就能建立直觉。这篇文章先从“慢在等待”这个朴素认知开始再带出开头的观点异步不是让单个请求变快而是让程序在等待时不闲着。理解了这件事剩下的都只是API细节。