做了三年Python开发干过爬虫、写过脚本、搭过服务但我真正理解Asyncio是在一次重写批量下载工具的时候。那会儿用requests循环抓几百个文件跑一趟得几分钟后来换成asyncio一分钟不到就完事。从那以后我就明白异步编程不是锦上添花是Python开发者迟早要跨过去的一道坎。这篇不是从官方文档翻译过来的“教程”是我在实际项目里摸出来的经验总结。我会用最直白的话讲清异步编程的核心原理、关键API和完整的落地案例最后整理一份高频报错的排查手册。不管你是刚学Python的新手还是写过同步代码但没玩过并发的进阶选手照着这篇文章的思路走一遍至少能在自己项目里用起来。1. 为什么非学不可异步编程到底解决什么问题1.1 同步代码的痛点从一次慢速请求说起先看一段很常见的同步代码import time import requests def fetch(url): print(f开始请求: {url}) resp requests.get(url) print(f完成: {resp.status_code}) return resp def main(): url https://httpbin.org/delay/2 # 这个接口会故意延迟2秒 start time.time() for i in range(3): fetch(url) print(f总耗时: {time.time() - start:.2f}s) main()跑一下耗时约6秒。问题很明显每次requests.get()发出后程序就卡在那里等服务器响应。等待期间CPU在干什么什么都没干就是空转等网络数据回来。换句话说这段时间你的程序是被网络IO阻塞住的。网络请求只是IO阻塞的一种文件读写、数据库查询、外部接口调用都是同一类问题。如果一个程序里有很多这样的操作同步写法就是眼睁睁地看着宝贵的时间一秒秒流走。1.2 asyncio的优势在哪里什么场景适用要解决等待浪费问题传统的做法是多线程。用线程池开几个线程让它们并行等。但Python有个绕不开的东西叫GIL同一时刻只能有一个线程执行Python字节码。好在对于网络IO这类操作在线程等待时GIL会释放所以多线程确实能提速。但线程也有代价线程的创建和切换有系统级开销线程多了还容易出竞态问题管理起来麻烦。asyncio走的是另一条路单线程靠协作式调度来切换任务。程序里只有一个事件循环在跑但它能记住“现在这个任务在等IO先让它歇着去执行另一个任务”等IO回来了再回来继续。这种切换开销极小轻轻松松管理成千上万个并发任务。不过也要说清楚适用边界。asyncio擅长的是IO密集型任务——网络请求、文件读写、数据库操作这类。如果是CPU密集型任务比如大量计算、图像处理asyncio帮不上忙那是多进程该干的活。常用三角形类比进程适合CPU密集线程适合阻塞型IO协程适合高并发IO各有各的相对优势。我个人的选择标准很简单——如果任务是等外部资源返回就先想想能不能用asyncio。提示Python 3.7才能完整使用我下面讲的asyncio.run()等新API老版本的话建议升级别拿兼容老版本当借口让自己痛苦。2. 事件循环与协程asyncio的两个核心概念2.1 事件循环到底怎么“转”起来的很多新手一上来就背概念说事件循环是负责调度协程的东西但没真正理解。我常用一个咖啡馆的比喻解释想象你是咖啡馆里唯一的服务员。同步模式下你接一个客人的单就非得等咖啡做出来端上去才去服务下一个客人。高峰期客人一多后面的客人全部干瞪眼。事件循环模式则是接到A客人的单记下来告诉后厨去做然后转头问B客人要喝什么B点完单你又去看看后厨的咖啡好了没有好了就端给A。你一个人但同一时间里服务了很多客人。这里的关键是“IO等待时切换到别的任务”。asyncio的事件循环本质上就是一个无限循环维护着两个数据结构——一个是”现在就执行“的任务队列一个是”在等某个事件回来“的等待队列。每次循环都检查有任务准备好继续了吗有的话就推进它的代码。所有准备好的任务都在这次循环的迭代里推进一小段。这种调度是协作式的意味着协程必须主动让出控制权。谁负责让就是你写的await。只有遇到await时当前协程才会把执行权交回事件循环事件循环才能去调度别的协程。如果不写await或者写了阻塞调用的代码那就等于在整个事件循环里插了一根铁棍谁也转不动。2.2 协程、Task、Future的关系这块我当年绕了很久现在用一张关系图给你理清楚不是绘图工具就是概念拓扑协程Coroutineasync def定义的函数不是协程函数被调用时返回的那个对象才是。它是一个待执行的计算流程但自己不跑得有人拉着它跑。Task任务把协程“注册”到事件循环上得到的对象。它才是事件循环真正调度的单元。你可以同时创建很多Task事件循环按状态推进它们。Task本质上有一个指向Future底层的机制任务完成的结果也会存在Future里。Future未来对象一个底层概念表示一个“将来才会有结果”的操作。你会见到它更多是因为第三方库自己写业务代码时一般只跟Task和协程打交道。一段代码看清楚三者的关系import asyncio async def say_hello(): await asyncio.sleep(1) return hello # 调用async函数 - 得到协程对象 coro say_hello() print(type(coro)) # class coroutine # 把协程包装成Task需在事件循环内 async def main(): task asyncio.create_task(say_hello()) print(type(task)) # class _asyncio.Task result await task print(result) # hello asyncio.run(main())新手最容易犯的错定义了async函数调用后以为它会自动执行。实际上你有三种方式让它真的跑起来——第一种直接await coro()第二种包成Task由事件循环调度第三种用asyncio.run()包一个最外层的入口协程。理解了这个区别你就能看懂下面所有的代码了。3. 入门必会的7个API与实操要点3.1 入口asyncio.run()的隐藏细节你写的异步程序从哪进来就是asyncio.run()asyncio.run(main())它做三件事创建新事件循环、把传入的协程跑完、最后关闭事件循环。我在第一次用的时候以为只要写个main()就能自动跑结果发现要手动调run()。它的一个特性是每次调用都会新起一个事件循环所以你不能在已有事件循环的环境里再调asyncio.run()。这点在Jupyter Notebook这种交互式环境里特别明显。你跑了一次asyncio.run(some_coro())再跑第二次时可能就报错asyncio.run() cannot be called from a running event loop——因为Notebook的kernel本身已经有一个事件循环在跑了。这时候我更建议你在Jupyter里用await some_coro()直接等待或者在脚本里用asyncio.run()保证整段入口干净。3.2 关键组合async/awaitasync def把一个普通函数变成协程函数。注意协程函数里的代码并不会像普通函数那样从上到下一次性执行碰到await就会暂停并让出控制权async def demo(): print(第一行) await asyncio.sleep(1) # 就在这里暂停让别的任务先跑 print(第二行)await后面必须跟一个“可等待对象”awaitable也就是协程、Task、Future三类之一。如果你await的不是这些东西会直接报TypeError: object int cant be used in await expression。另一个高频坑是在非async函数普通同步函数里使用await解释器会直接报SyntaxError: await outside async function。想要这个能力就把那个同步函数也改成async def。3.3 任务编排create_task、gather、wait怎么选如果只是单协程里加点延时用不上asyncio。真正的威力是并发创建一大堆任务。三个最常用API的差别asyncio.create_task(coro)把协程包装成Task并立即排入事件循环。它不等待结果返回Task对象。适合你要“发任务但不立刻等它”的场景。asyncio.gather(*coros, return_exceptionsFalse)同时执行多个可等待对象等全部完成后把结果按顺序返回。return_exceptionsTrue时某个协程抛异常不会影响其他任务继续执行异常会作为结果返回。这是最推荐直接用的方式。asyncio.wait(tasks, timeoutNone, return_when...)接收的是Task的集合返回(done, pending)两个集合。适合需要手动管理“某些任务还没完成”的场景比如设置统一超时。直观对比一下gather和create_task组合await的区别很多人困惑“结果顺序”。用gather你拿到的结果列表顺序永远和你传入协程的顺序一致不管谁先完成。用create_task分别await则谁先拿到结果由实际完成时间决定。如果业务需要“按请求顺序取结果”gather更省心如果需求是“有一个完成了就立刻处理”那建议用asyncio.as_completed迭代。配一个例子import asyncio async def fetch(num, delay): await asyncio.sleep(delay) return f第{num}个任务 async def main(): tasks [ fetch(1, 2), # 最慢的放前面 fetch(2, 0.5), fetch(3, 0.1) ] results await asyncio.gather(*tasks) print(results) # [第1个任务, 第2个任务, 第3个任务], 顺序不乱 asyncio.run(main())3.4 超时与取消别让卡死的请求拖垮程序异步任务最大的隐患就是“无限等待”。网络请求可能卡死数据库连接可能不返回没有超时控制的任务会占着坑不挪窝。asyncio.wait_for(awaitable, timeout)就是干这个的import asyncio async def slow_operation(): await asyncio.sleep(10) return 成功 async def main(): try: result await asyncio.wait_for(slow_operation(), timeout3) print(result) except asyncio.TimeoutError: print(3秒没完成超时了) asyncio.run(main())wait_for在超时时会自动取消内部的任务。而如果任务本身拒绝响应取消比如它正在执行一段无法被打断的同步阻塞代码超时也不会立刻生效——这个我们在第3.6节讲怎么处理。还有一种主动取消的方式直接调用task.cancel()让协程在下一个await点上抛出CancelledError。实际项目中我习惯给所有网络请求都套一层wait_for避免某个下游服务把整个程序拖死。这是我从一次线上故障里学到的极端经验一个没有超时的脚本等了一个第三方API 40分钟没有返回日志一片空白始作俑者就是没设置超时。3.5 限流Semaphore控制并发上限并发不是越大越好。你有1000个任务一口气全发出去目标服务器可能直接拒连接甚至把你的IP封掉。Semaphore信号量就是用来给并发加塞子的。import asyncio sem asyncio.Semaphore(5) # 最多同时5个 async def safe_fetch(url): async with sem: # 进来占一个名额 print(f抓取: {url}) await asyncio.sleep(1) return f结果: {url} async def main(): tasks [safe_fetch(fhttps://api.example.com/{i}) for i in range(20)] await asyncio.gather(*tasks) asyncio.run(main())注意它的写法是**async with sem:**不是普通的with sem。如果写错了会报AttributeError: __aenter__。限流参数怎么定我的经验是先看目标服务的承载能力不确定就测试性地从5开始逐步往上调。爬虫场景下这个参数同时是礼貌与生存的边界。3.6 阻塞调用怎么办run_in_executor把同步函数交给线程池这是asyncio的“兜底手段”。上面强调过协程里如果出现同步阻塞调用比如requests.get()、time.sleep()、普通的open()文件IO整个事件循环会被卡死所有并发任务全部停摆。然而实际项目中总免不了要用些同步的库比如还没出异步版本的SDK。解决办法是loop.run_in_executor()它把这些阻塞调用丢给一个线程池去执行返回的Future可以被awaitimport asyncio import requests def sync_fetch(url): # 这是一个普通的同步函数 resp requests.get(url, timeout5) return resp.status_code async def async_wrapper(url): loop asyncio.get_running_loop() # 丢进默认线程池执行不会阻塞事件循环 result await loop.run_in_executor(None, sync_fetch, url) return result async def main(): urls [https://httpbin.org/delay/1, https://httpbin.org/delay/2] results await asyncio.gather(*(async_wrapper(u) for u in urls)) print(results) asyncio.run(main())也可以显式传一个ThreadPoolExecutor控制线程池大小。注意不要滥用run_in_executor它的原理是线程用多了又回到线程的老路上去。主要用在你无法替换同步库的场景而像requests这种完全可以换成aiohttp就换成aiohttp。4. 从零能跑一个完整的并发爬虫案例4.1 需求与代码骨架光讲API不落地没有意义我们来做一个真实场景。假设要抓取50个网页的标题用同步方式挨个抓每个页面延迟1秒就要50秒。换个异步思路事件循环同时发起50个请求50个请求的总耗时逼近最慢的单请求也就是约1秒。先装上需要用到的库环境是Python 3.7建议装到虚拟环境里pip install aiohttp核心代码结构是这样的import asyncio import aiohttp import time async def fetch_title(session, url): try: async with session.get(url, timeout10) as resp: html await resp.text() # 简单提取title标签内容 start html.find(title) len(title) end html.find(/title) title html[start:end].strip() if start ! -1 and end ! -1 else 无标题 return url, title except Exception as e: return url, f错误: {e} async def main(): urls [fhttps://httpbin.org/delay/1 for _ in range(50)] # 模拟50个耗时网页 async with aiohttp.ClientSession() as session: tasks [fetch_title(session, url) for url in urls] start time.time() results await asyncio.gather(*tasks) print(f总耗时: {time.time() - start:.2f}s) for url, title in results[:3]: print(f{url} - {title}) asyncio.run(main())重点在async with aiohttp.ClientSession()。整个程序共享一个Session连接会被复用效率远高于每次请求都建新连接。你的所有请求都包成Task交给gather去调度。4.2 同步版本 vs 异步版本效果对比我拿上面的例子在本地跑过50个请求、每个延迟1秒同步版requests逐个请求约50.3秒异步版aiohttp并发请求约1.8秒差了接近28倍。这还只是在本地模拟环境下真实网络场景的抖动会更明显。需要说明的是很多新手拿这个对比后兴奋过头把asyncio当成万能加速器——不是的。你换成CPU密集型的计算试试比如大量循环运算或者数据处理——那种任务用asyncio不会快因为计算本身占据CPU时间片协程切换反而有损耗。asyncio的快来自“等待IO时不占资源”这个原理要真正刻在心里。4.3 进阶加超时、限流、异常兜底真实环境里不能就这么裸奔必须做三重防护第一重请求超时。session.get(url, timeout10)不能省避免某个服务器挂掉后疯狂占用连接。**第二重并发限流。**前面爬相同网站的接口很容易被反爬。给fetch_title加一个信号量SEM_LIMIT asyncio.Semaphore(10) # 最多10个并发 async def fetch_title(session, url): async with SEM_LIMIT: try: async with session.get(url, timeout10) as resp: html await resp.text() return url, html except asyncio.TimeoutError: return url, 超时 except aiohttp.ClientError as e: return url, f网络错误: {e}第三重结果容错。gather里某个任务失败不能让整个程序崩溃。上面的代码已经在函数内部try/except了所以安全。如果你不想在函数内部处理也可以results await asyncio.gather(*tasks, return_exceptionsTrue) # 拿到结果后筛一遍except Exception类型的结果就是失败任务这三重防护加完后再去看全量代码你会发现整体结构没有变得更复杂——这就是asyncio的魅力代码看起来是同步顺序写的顺着纸面从上往下读实际执行时是并发的。5. 常见问题与排查技巧实录5.1 三个高频报错的定位与解决我整理了一张自己在实际项目中踩过的坑排查表基本都是排在最前面的高频报错报错信息出现场景原因解决方法SyntaxError: await outside async function在普通def函数里写了awaitawait只能在async def函数体内使用把外层函数改为async def或者把相关逻辑移到协程函数里RuntimeError: asyncio.run() cannot be called from a running event loop在Jupyter Notebook等交互环境或在一个协程内部调用asyncio.run()当前线程已经有事件循环在跑了而asyncio.run()要求新建一个交互环境直接用await协程内要启动子任务就用asyncio.create_task()不要用run()RuntimeWarning: coroutine xxx was never awaited调用了async函数但没有await它协程对象没有执行机会被垃圾回收时提示确认调用处没有漏写await如果是要“后台溜任务”用create_task()包起来RuntimeError: Event loop is closed循环内资源未清理或多次使用asyncio.run()后对象还引用旧循环事件循环已被关闭但某些Task或回调还挂着用asyncio.run()统一管理生命周期需长期运行的任务考虑asyncio.new_event_loop()并显式set_event_loop()TimeoutErrorasyncio.TimeoutErrorwait_for到了指定时间还没完成超时后任务被取消或抛出异常捕获后做降级处理给下游服务提示排查目标服务可能已假死最后这个Event loop is closed是最容易在复杂的项目里抽风的。之前我在一个用Quartasyncio版Flask写的服务里数据库连接池在应用关闭时没释放干净重启就报这个错排查半天才找到是连接池的引用没销毁。建议排查时先看“哪些资源还在使用旧事件循环”连接、子进程、信号处理都会触发。5.2 调试与测试的独家心得调试异步代码比同步代码难在“时序不可控”。断点停在某个协程里时其他协程还在跑日志交错在一起非常容易看晕。分享几个我长期在用的技巧**第一日志里带上任务ID。**给每个协程传一个唯一标识比如URL、编号输出时始终带上否则日志里全是交错行没法判断哪一步对哪一步。**第二利用loop.slow_callback_duration或者代码内自己测量任务耗时。**我在包协程时习惯用装饰器记录每个任务耗时方便对比哪个阶段是性能瓶颈。**第三测试异步代码用pytest-asyncio别自己搞asyncio.run套浏览器。**这个插件支持给测试函数加标记写起来还很简洁import pytest pytest.mark.asyncio async def test_fetch_title(): async with aiohttp.ClientSession() as session: url, title await fetch_title(session, https://example.com) assert title第四调试时把asyncio.get_event_loop()换成asyncio.get_running_loop()。这是Python 3.10以后被反复敲打的一个变化老代码里到处是get_event_loop()在新版本下会收到DeprecationWarning。5.3 三个容易犯的“新手级”逻辑误区除了报错我更想多唠叨几个“代码能跑但设计错误”的误区。这些在项目评审里我可没少批评小朋友。误区一在协程里用time.sleep()。这个说到天荒地老也要强调它会让当前线程睡过去事件循环整个停摆所有并发任务全部冻结。必须用await asyncio.sleep()。误区二把asyncio.open_connection()或aiohttp等库用于CPU密集任务。比如你会写“我用async爬了1000个网页然后用BeautifulSoup解析”但解析是CPU计算这段会阻塞事件循环。数据量一大该切出去还是得切出去比如用run_in_executor。误区三用gather开10000个任务不设限。异步开销小归小但任务对象本身也有资源成本一次性创建上万个任务会导致内存飙升。配合Semaphore限流才是工程级用法。结束语我的实战经验总结如果你完整看完上面这些内容现在应该有能力把一段同步的IO密集代码改写成asyncio版本了。我给自己的项目做代码评审时碰到IO密集场景会下意识问一句——这里能不能异步化但也会立刻反问——值得不值得有时候任务量很小一个同步requests.get()就搞定强行引入asyncio徒增复杂度那种场景“同步到底顺手清晰”反而是最好的方案。最后分享一个我在线上项目里常用的可靠小技巧对异步任务统一封装一个“限时重试熔断”的函数不要每个业务各写各的。脚手架搭好后后续接任何一个第三方接口都只需要写业务逻辑超时和重试不用再重复考虑。具体而言重试逻辑用一个简单的循环包在wait_for外面重试间隔用asyncio.sleep这样整个重试过程不会阻塞主流程。异步编程的上手曲线确实比普通Python语法陡一点。但一旦理解了事件循环、协程、Task这三个概念的协作关系你写出来的程序在IO密集场景下的提升是肉眼可见的。如果这篇文章帮你减少了一点点试错时间那它就是有价值的——接下来拿自己手头最耗时的那个脚本开刀改成异步试试。