深入理解 Python asyncio：事件循环、协程与任务调度

开头聊到 Python 异步编程绕不开的就是asyncio。这套标准库从 Python 3.4 开始进入视野到 3.7 之后语法基本定型算是目前 Python 生态系统里做并发 IO 的基石之一。很多刚接触的朋友一翻开文档看到async def、await、event loop、coroutine这一堆概念第一反应都是“这到底解决了什么问题我和普通函数有什么区别”。这篇文章就从我实际写爬虫、做批量请求、写小工具服务这几年的经验出发把asyncio的脉络捋清楚它帮你解决了什么痛点、核心对象是什么、真正的使用套路是怎样的、以及坑都藏在哪儿。无论你之前只写同步代码还是已经写过一点但老是踩到“事件循环已关闭”“协程未等待”这类报错这篇都适合当作一本带注释的入门手册来读。我会从最基础的设计思路开始讲把event loop、coroutine、task、future这几个词放到真实场景里解释然后给出一套可以直接抄作业的代码骨架再列出我实际调试中反复遇到的高频问题。这是一篇务实的操作笔记不是官方文档的复读机所以你最好跟着敲一遍只靠看容易“眼睛会了手不会”。1. 内容整体设计与思路拆解1.1 异步到底解决的是什么问题很多人被“异步比同步快”这句话带偏了方向以为异步就是无脑加速。实际上asyncio对 CPU 密集型任务毫无办法它真正擅长的是 IO 密集型场景。判断标准很简单如果你的程序卡住是因为等网络响应、等磁盘读写、等数据库返回而不是在疯狂做计算那么异步就能帮你把排队等待的时间腾出来干别的活。举个例子你用requests写一个脚本去抓 100 个网页同步逻辑就是一个一个来请求发出去之后线程就发呆等响应假设每个请求平均 200 毫秒100 个就是 20 秒起步。换成asyncio配合aiohttp之后你可以同时发出 100 个请求总耗时还是大约 200 到 300 毫秒左右。这不是魔法而是底层让网卡和操作系统帮你同时维护多条连接程序在等待某一条连接返回期间立刻切换到另一条连接去处理它已经到达的数据。这里有一个关键认知单线程照样可以做并发。传统多线程是操作系统帮你频繁切换线程asyncio是程序自己在同一个线程内手动切换任务。你能控制切换点反而比操作系统盲抢时间片更可控也不会为了解锁共享变量而争得头破血流。1.2 为什么选了 asyncio 而不是多线程或协程库初学者最常见的纠结就是在threading、multiprocessing、asyncio三兄弟之间做选择。我的建议是在纯 Python 的 IO 密集型场景里优先考虑asyncio理由其实很现实线程切换开销和内存占用不小。每个线程默认栈空间就好几 MB开 1000 个线程你光内存就扛不住。多线程的共享变量加锁问题非常分散精力。你要处理Lock、RLock、Condition、Queue之间的配合稍微乱一点就是死锁或者数据竞争。进程模型适合 CPU 密集型但进程间通信成本高不适合大量轻量化 IO 任务。asyncio自己的调度是在一个线程内完成的切换任务的开销极小你能随手创建上万个协程都不心疼。当然如果你遇到的是某个第三方库内部阻塞了比如很多同步数据库驱动、老的文件操作库它们不会主动让出控制权那么asyncio也救不了你。这时候方案是两条路要么换协程版驱动比如把requests换成aiohttp把原生的文件读写放到asyncio.to_thread或run_in_executor里要么老老实实上多线程。搞清楚这个边界后面就不会上来就喷“异步没用”。1.3 新手方案选型协程、任务、事件循环一起用真正写业务代码时你不会手动去操作底层那些Future而是站在更高层使用两个东西asyncio.run()负责启动宇宙asyncio.create_task()负责创造调度单位。协程函数用async def定义调用一个协程函数并不会立刻执行函数体而是返回一个协程对象。你需要让它进入事件循环它才会跑起来。我建议新手把事件循环理解成一个外卖派单台你接到 100 个订单骑手只有一个人他送完一份再送下一份但他会在等商家出餐的间隙去取附近已经做好的餐。asyncio.run就是“今天开始营业”create_task就是“这个订单列入派送计划”await就是“当前这个订单没出餐我先去干别的”。这套心智模型一旦建立起来你再看任何asyncio代码都不会觉得玄。官方文档里那些类图、状态转移图本质上都是在描述派单台内部如何调度你完全可以在第一遍学习时忽略它们先把“启动-注册-等待”这三步操作练熟。2. 核心细节解析与实操要点2.1 认识 async def、await 和协程对象所有asyncio代码的基础语法就三行async def my_coro(): await asyncio.sleep(1) return done一个函数只要用async def声明它就不再是普通函数。调用my_coro()时你不会得到返回值而是得到一个提醒你“这是一个尚未运行的协程对象”的东西。在 Python 3.12 之前如果你忘了把它放进事件循环解释器还会给你打一条RuntimeWarning: coroutine was never awaited警告很多人第一次看到这玩意儿都一脸懵。await关键字表示“我在这里让出控制权等待某个可等待对象完成”。可等待对象主要有三类协程对象、Task对象、Future对象。日常使用时 90% 的时候你await的是一个协程对象或 Task。注意await只能在async def函数体里使用在普通函数里用关键字语法就直接报语法错误。这里有个初学者必踩的小坑以为用await就会立刻执行完其实await只是挂起等待不保证马上执行。比如你await asyncio.sleep(0)它也是先挂起等事件循环下一轮再唤醒。理解这一点后面排查“为什么我明明 await 了顺序还是不对”时会少很多痛苦。2.2 事件循环怎么帮你“凭空”调度事件循环是整个asyncio引擎的心脏它在每个时刻维护一个“就绪队列”和一些“等待队列”。程序在某个协程里执行到await时会把当前协程挂起并把要等待的事件注册进事件循环。比如asyncio.sleep(1)会让事件循环记一笔一秒后唤醒这个协程。然后事件循环立刻从就绪队列里取出下一个协程来跑。这个过程听起来复杂但你几乎不需要手动直接调用事件循环对象的方法除非你在写框架或维护老代码。官方现在推荐的最高层入口是asyncio.run(main())asyncio.run会帮你创建一个新的事件循环运行给定的协程等它结束后关闭事件循环。你不需要写loop asyncio.get_event_loop()也不需要最后调用loop.close()这些容易被搞错的生命周期管理asyncio.run一次性包办。如果你在 Python 3.7 之后的版本里写代码请默认用这个入口不要用老教程里的loop.run_until_complete那套写法容易在现代环境下碰见“事件循环已关闭”的毛病。值得注意的是一个asyncio.run调用内部只存在一个事件循环实例你在这个入口之外创建的任务不会自动进入这个循环。所以我写程序时习惯把所有并发逻辑集中到main()协程里不让业务逻辑散落在模块各处。2.3 Task 是并发的最小调度单位光有协程对象还不够如果你直接写coro my_coro() await coro那只是顺序等待一个协程完成并不算真正的并发。要想并发你需要把协程包装成Task也就是“已经提交给事件循环安排执行”的对象async def main(): task1 asyncio.create_task(my_coro(first)) task2 asyncio.create_task(my_coro(second)) await asyncio.gather(task1, task2)create_task调用后协程立刻被登记到事件循环中排队你不用立刻await它之后任意时刻都可以await task来获取它的结果或等待它完成。这就是“并发”的操作入口。 如果你创建了 Task 但到最后都没await它程序结束时会收到警告因为 Task 被“遗弃”了。日常写法里最优雅的并发汇总方式是asyncio.gather(*tasks)它会同时等待所有 Task 完成并按传入顺序返回结果列表。还有一个细节我经常提醒自己gather虽然好写但如果你想要“谁先完成就先处理谁”的效果它不合适。这时候要换asyncio.as_completed()它返回一个迭代器每完成一个 Future 就吐出一个结果适合做动态进度处理和流式输出场景。2.4 锁、信号量和队列协程之间也要沟通协程虽然不共享线程栈但可能共享全局变量或同一个资源比如同一个文件、同一个第三方接口的限流配额。这时候需要asyncio.Lock、asyncio.Semaphore、asyncio.Queue。用法上和多线程版本很相似但有个致命区别这里的锁不能用普通的with lock:语法因为with不支持挂起必须用async with lock:。比锁更常用的是信号量用来限制并发规模。比如你抓 1000 个页面但目标服务器只扛得住 20 个并发于是写semaphore asyncio.Semaphore(20) async def fetch_one(url): async with semaphore: return await aiohttp.request(GET, url)这样能保证任意时刻最多 20 个请求在跑避免把自己的 IP 送进小黑屋。这个写法是异步爬虫里最核心的限速手段我再三强调千万不要小看它很多“异步反而把服务器打挂了”的事故就是因为没加信号量。队列则用于生产者消费者模型。生产者负责把任务塞进asyncio.Queue消费者负责从队列里取任务执行。它自带线程安全属性你在多个协程间安全传递消息不需要额外加锁。3. 实操过程与核心环节实现3.1 一个完整的“几十个请求并发抓取”Demo这里我写一个可以原封不动拿过去跑的最小例子。假设我们要抓取一组 HTTP 接口每个接口模拟耗时一秒最终打印每个接口的状态码。完整代码如下import asyncio import aiohttp async def fetch_one(session, url): async with session.get(url) as resp: return url, resp.status async def main(): urls [fhttps://httpbin.org/delay/1?num{i} for i in range(50)] semaphore asyncio.Semaphore(10) async with aiohttp.ClientSession() as session: async def bounded_fetch(url): async with semaphore: return await fetch_one(session, url) tasks [asyncio.create_task(bounded_fetch(url)) for url in urls] results await asyncio.gather(*tasks) for url, status in results: print(url, status) if __name__ __main__: asyncio.run(main())如果换成同步requests这段代码跑完大约需要 50 秒因为每个请求等 1 秒。而这里受信号量 10 的限制并发批次是 50 除以 10约等于 5 批每批 1 秒总耗时也就 5 秒左右。加上网络波动撑死 6 秒。这就是最直观的效率对比你把等待时间从“串行累加”压缩成了“分批并行”。这里有个容易被忽略的点ClientSession必须使用async with上下文管理器创建因为它在内部维护连接池使用完要异步关闭不能像requests.Session那样只调close()。3.2 超时控制别让你的程序永远在等异步程序一旦出现某个第三方接口不响应整个await就会一直挂下去因为你没有给协程设置超时机制。很多新手自己跑例子跑得好好的一上真实环境就“程序卡死”八成是漏了这一环。控制超时有两个通用姿势。第一个是用asyncio.wait_for写法非常直观try: result await asyncio.wait_for(fetch_one(session, url), timeout3) except asyncio.TimeoutError: print(f{url} 超时了)第二个是用上下文管理器asyncio.timeout在 Python 3.11 之后可用更适合给一段代码整体设限async with asyncio.timeout(5): result1 await fetch_one(session, url1) result2 await fetch_one(session, url2)给每个外部请求加上超时是我写异步代码的铁律。但不建议超时设太短否则网络稍微抖动就误杀正常请求。我一般的经验是普通网页请求超时给 10 秒内部 API 调用给 3 到 5 秒文件上传这类的按体积另行评估而不是统一用一个拍脑袋的数字。3.3 分批任务调度避免一次性创建一万个任务有些人以为create_task越猛越好上来就给每个 URL 创建一个 Task10000 个请求瞬间全部注册。这在小型 demo 上没事但遇到IP 限流、文件描述符上限、内存吃紧时就会出问题。合理的做法是“信号量限流 动态创建任务”而不是把任务全塞进列表再gather。一个更健壮的批量调度姿势是用 worker 模式固定创建几个 worker 协程每个 worker 不停从队列里取任务队列空了就退出。async def worker(q, session): while True: try: url await asyncio.wait_for(q.get(), timeout0.5) except asyncio.TimeoutError: break await fetch_one(session, url) async def main(): q asyncio.Queue() for url in urls: await q.put(url) workers [asyncio.create_task(worker(q, session)) for _ in range(10)] await asyncio.gather(*workers)这种模型的好处是任务不是一次性全部进入事件循环而是按需取用worker 数量就是并发上限。你不需要同时维护 10000 个 Task内存和心理负担都小很多。真实的生产脚本里我绝大多数时候都采用这种队列加 worker 的模式只有临时测试才直接用列表推导式一把梭。3.4 协程间结果回收与异常处理并发任务有一个容易让人困惑的点如果其中一个 Task 抛异常gather的默认行为会导致整个await抛错其他任务继续归其他任务继续跑还是会被取消行为可能和预期不一致。Python 3.11 前的gather一旦有子任务异常整个等待点就炸了未完成的任务会留在事件循环里。新版提供了gather(..., return_exceptionsTrue)可以让异常被当作结果返回而不影响其他任务。所以我在需要逐任务处理结果时会主动给每个任务包一层“异常消化”函数async def safe_fetch(url): try: return url, await fetch_one(session, url), None except Exception as e: return url, None, e这样在结果层面统一检查第三个元素是不是None处理逻辑清晰很多。后期的日志分析也方便不会只看到一条 Traceback 然后整个任务列表消失。顺带提一下取消机制。Python 里调用task.cancel()会在目标协程的await处抛出asyncio.CancelledError。如果某个协程需要做清理工作比如关闭子资源、回写状态可以在except asyncio.CancelledError里处理后再重新抛出。这个机制写框架时躲不开但普通业务脚本里用得少知道有这么回事即可。4. 常见问题与排查技巧实录4.1 Top 5 高频报错排查速查表我把这几年带人入门时被问得最多的几个报错整理成表每条都附上直接解决方案。报错信息根因排查/解决RuntimeWarning: coroutine was never awaited调用了async def函数但未把它交给事件循环检查是否漏了await或者在普通函数里调用协程要么改成await coro()要么用asyncio.create_task(coro())RuntimeError: asyncio.run() cannot be called from a running event loop在事件循环内部又调用了asyncio.run在async函数里不要再调asyncio.run直接把协程await掉asyncio.run只能作为根入口RuntimeError: Event loop is closed事件循环被提前关闭任务还在里面检查是否手动调用了loop.close()在asyncio.run内部不要主动去获取并关闭get_event_loop()的循环TimeoutError或连接卡死没有设置超时外部服务无响应用asyncio.wait_for包装每个可等待操作确认网络与目标服务可达ValueError: too many file descriptors并发请求数超过系统文件描述符限制降低并发量使用Semaphore控制或者调高系统ulimit -n但优先还是限流这里特别要说一句RuntimeError: asyncio.run() cannot be called from a running event loop在新手问的问题里常年霸榜很多人的代码是两个协程互相调用在每个协程里都想用asyncio.run启动子协程正确做法是直接await子协程而不是重新启动一个新循环。4.2 为什么我的“并发”还是慢得像同步最常见的原因是你在协程里调用了同步阻塞库比如直接用了requests.get()。假设你写了async def fetch_data(): return requests.get(url).text那么当协程运行到这一行时整个线程都会被卡住因为requests.get是同步阻塞调用它不会通知事件循环“你去找别人吧”而是老老实实占用线程等网络响应。这样一来无论你开多少个 Task任务依然是一个接一个阻塞着执行异步的优势直接归零。解决方案就是换车HTTP 用aiohttp或httpx的异步接口数据库用对应驱动的异步版本文件读写可以用asyncio.to_thread(fn, *args)把阻塞操作丢到线程池执行。import asyncio def read_large_file(path): with open(path, r) as f: return f.read() async def main(): data await asyncio.to_thread(read_large_file, /tmp/huge.txt) print(len(data))asyncio.to_thread是 Python 3.9 引入的便利函数它内部会自动使用默认的执行器线程池适合把少量、低频的阻塞操作化解放进异步流程中。注意别用它处理高频操作否则线程池会变成新的瓶颈。4.3 排查死锁与“程序退出但任务没跑完”协程死锁不像多线程死锁那么显眼因为事件循环会把控制权交来交去你很难直接从堆栈里看到两个线程互相等待。常见协程死锁场景有两种。一种是两个协程互相 await 对方完成即循环等待另一种是在某个协程里用阻塞循环等待某个 Task 的状态比如写while not task.done(): pass这会导致事件循环没有机会去推进那个 Task。排查手段很简单给事件循环设置调试模式asyncio.run(main(), debugTrue)开启调试模式后事件循环会记录每个协程的创建位置、当前等待点如果某个协程超过 100 毫秒没有被推进它会打印警告信息和堆栈。这对于定位“哪个协程卡在哪个 await 上”非常高效。我在本地调试偶发卡顿问题时几乎必开这个开关。另外程序执行完但子任务没跑完多半是因为你用了asyncio.run(main())但main()里没有等待所有 Task 完成。比如创建了 Task 列表却不await gather它们main()跑完事件循环关闭剩下的 Task 直接被取消。所以规范做法永远是创建任务后要确保它们被一个或多个await收集到结果或者在异常时被取消绝不能只创建不消费。4.4 生产环境独家避坑建议如果脚本要部署到服务器上长期跑有几个坑我强烈建议提前避开不要用asyncio.sleep()做精确计时。它最小精度受系统时钟影响且调度顺序复杂时误差可能累计。要精确定时就维护一个“下次执行时间”变量动态计算剩余等待时间。日志里打印协程状态时不要直接print(task)普通print出来的只是一堆对象地址。要用task.get_name()task.get_coro()或直接repr(task)来看它挂起在哪个调用点。不要把asyncio.Queue.get()放在无保护的while True里空闲时会无限空转。要么设超时要么在队列尾部放一个“哨兵值”告诉 worker 结束。如果脚本涉及长时间运行建议定期重启进程因为事件循环在运行多天后可能遇到底层 socket 描述符泄漏或者第三方库内部资源累积。这是很多跑监控脚本的同行踩过的坑靠代码只能缓解不能根除。写压力测试时记得把信号量、超时、重试三件套都做进被测接口否则你会把真实接口测挂或把日志刷爆。5. 进阶扩展从入门到能独立设计异步服务入门之后很多人的下一步需求是写一个完整的异步服务而不仅仅是脚本。此时需要掌握三个方向异步上下文管理、异步迭代器、以及把异步代码嵌入同步工程。异步上下文管理器就是实现__aenter__和__aexit__方法的类支持async with语法。你如果自己封装一个数据库连接池或者客户端对象就应该实现这两个方法这样资源释放逻辑统一而不容易漏。异步迭代器则是实现__aiter__和__anext__的对象支持async for。比如你想分批从数据库取数据每取一批需要异步等待 IO 返回就可以用异步迭代器把“取下一批”这个动作包装得和普通for一样干净。至于在同步工程里引入异步常见的做法是单独起一个后台线程并在线程内调用asyncio.run()这样同步主流程可以随时向异步事件循环提交任务。但要注意这个模式里异步事件循环的线程非常容易被自动化测试里反复重启搞乱最好是把这个线程的生命周期封装成一个完整的管理类而不是散落在任意代码里创建事件循环。到了这个阶段你再去回头读官方文档里关于loop、Protocol、Transport的章节就会轻松很多。因为底层那些回调机制不过是你用create_task时帮你做好的事情只是框架或者高性能场景需要你直接操作它们。就我个人体验而言asyncio最大的坎不是语法而是思维模式的转变不要用“顺序执行”去理解代码而是想象你同时在等待许多条河里的鱼上钩哪条先动就先拉哪条。一旦建立的模型对了看什么示例都能迅速对应到“启动、注册、等待、回收”这四个动作上。如果这篇文章能帮你少走一点弯路我会很高兴。最后再分享一个小技巧写异步代码时可以把debugTrue长期留在本地开发的启动入口里它会用一点点性能开销换来清晰的调度日志等你真正上线再关掉也不迟。