1. 先建立正确的直觉：事件循环到底是个什么东西
在聊 asyncio 之前，我建议你先忘掉“并发”、“异步”这些让人头皮发麻的词。你就把 Python 解释器想象成一个干活的工人，这个工人一次只能干一件事，不能同时左手画圆右手画方。但生活里会有很多“等待”的事情：等文件从磁盘读完、等网络数据包回来、等数据库返回查询结果。在同步代码里，这种等待是傻等，CPU 空转，工人把手上的活停住，盯着钟表发呆。而 asyncio 里的“事件循环”——也就是 Event Loop——本质上就是这工人手里的一个“智能待办清单”，它让工人在等 A 事情的时候，赶紧转头去干 B 事情的活，等 A 的结果回来了，再回来继续处理 A。
事件循环（Event Loop）是 asyncio 最核心的调度中枢。它的名字已经说明了它的工作方式：一个死循环，不断在“检查有哪些任务可以执行”“执行它们”“检查它们是否完成”“把完成的结果分发出去”这几件事之间转圈。
我用一个特别接地气的例子来说明，假设你是个外卖配送员，一个人在一条街上送餐。同步模式是：你拎着 A 订单的餐，站在客户楼下等，客户不下来你就一直等着；等 A 送完了，再去 B 供应商取餐，再站楼下等。asyncio 模式是：你发现 A 客户楼下门禁要等，就把 A 的餐先挂在车把手上，掉头去 B 供应商那边取餐，B 还没做好，就先回到 A 楼下看一眼；谁准备好了就先处理谁，你的时间一直在“取餐、送餐、确认状态”这个循环里流动。事件循环在这里就是“你的个人调度系统”，它负责回答一个核心问题：当前时刻，哪件事最有资格被推进？
所以，理解事件循环最关键的心智模型是：单线程 + 协作式多任务。单线程意味着没有锁竞争、没有线程切换开销、没有 GIL 抢来抢去的破事；协作式意味着每个任务必须主动让出控制权（yield），让事件循环有机会去安排别的任务，而不是操作系统强杀式地抢占。如果某个任务死活不交出控制权，那事件循环就被卡住，其他所有任务全部停滞——这就是初学者最容易踩的坑，我在第 3 节里会专门讲。
适合看这篇内容的人，我说直白一点：你已经写了几段 async/await 代码，跑通了官方文档里的例子，或者你在网上抄了别人的异步爬虫代码，能用但似懂非懂，一遇到各种循环报错就慌了。这篇内容的目的就是帮你把最后那层窗户纸捅破，之后你再看 asyncio 的源码、看别人的异步框架，思路会顺畅很多。
2. 事件循环的底层运转机制：它靠什么转起来
2.1 事件循环的三个核心组成部分
一条真正在运行的事件循环，背后会有这么几块东西，我先帮你拆开：
待执行任务队列（Ready Queue）：所有已经准备好可以被执行的协程任务，都会挂在这个队列里。事件循环在每个 tick（一次循环迭代）中会从里面取任务依次运行。注意这里的“就绪”不一定是指任务完成了什么复杂的计算，更多的可能是它所等待的 IO 已经就绪了，或者它已经被某个回调唤醒了。
等待中的 Future/Task 表：很多协程执行到某一步会挂起，比如await asyncio.sleep(1)，这时事件循环不知道它什么时候醒来，就会把它登记在自己的“沉睡名单”里。事件循环每转一圈都去看看这个名单，谁的等待时间到了、谁等待的文件描述符可读了，谁就会被挪到 Ready Queue 里去。
事件监听器（Event/IO Poller）：这是最底层的东西，在 Linux 上是epoll，在 Windows 上是IOCP，在 macOS 上是kqueue。事件循环不光要管“时间到了”这种基于时钟的任务，还要管基于网络 socket 的 IO 事件。这里需要说一下，asyncio 的主要用途其实是 IO 密集型任务，它需要用底层的系统调用去问操作系统：这个 socket 有数据了吗？那个 socket 可以写了吗？等你理解到这一层，你才会明白为什么事件循环被称为“循环”，它不是在转圈碰运气，它是在反复问操作系统“外部世界的状态变化了没有”。
这三者配合起来，事件循环的一个完整迭代过程就像这样：
- 调用 epoll/kqueue/IOCP 等系统调用，询问“我关心的这些 fd（文件描述符）有没有事件发生”，可以设置 timeout，这个 timeout 是选最近的定时器时间。
- 如果内核说有事件发生，就把相应的回调或者 Future 标记为就绪。
- 把到期的定时器任务也标记为就绪。
- 遍历 Ready Queue，逐个运行它们的回调代码。
- 回到第 1 步，继续循环。
2.2 从协程到 Future：事件循环如何知道任务在等待
这里就涉及经典问题：asyncio.sleep(1)里面到底发生了什么？为什么它能让出控制权？
在 Python 中，一个协程函数（用 async def 定义）被调用时并不会真正执行函数体，而是产生一个协程对象。当你把协程对象包装成 Task 并交给事件循环之后，事件循环会反复调用这个 Task 的__step方法。协程对象在执行的时候，一旦碰到await，它就会暂停，并把一个“下一步要做什么”的指示返还给 Task。
举个例子，await asyncio.sleep(1)这个表达式的妙处是：它内部会创建一个 Future（或者更准确地说，一个_asyncio.Future），然后调用事件循环的call_later，让事件循环在 1 秒之后帮它 Future 设置结果。然后协程直接就yield了，把“我正在等这个 Future，它完成了叫我”的信息反馈给外层 Task。Task 拿到这个信息后，就对这个 Future 注册一个回调：等你的结果出来，就把我再次放进就绪队列里。之后，Task 自己的执行就暂时告一段落。
关键点来了：如果这个 Task 明明在等待，却依然待在事件循环的主循环里，那其他任务就没法执行了。所以事件循环必须把这种处于等待状态的任务“摘”出去。事件循环其实并不完全是被动地“轮询”所有任务，它更像一个传令兵：谁完成了、谁被唤醒了，谁才进入下一轮的候选名单。
换个生活化的比喻，协程就像是你在公司里同时发了好几封邮件给不同部门，每封邮件就是一个任务。你先把邮件都发出去（这就是进入事件循环），你不需要每封邮件都等对方回复了再发下一封。然后你的同事（事件循环）帮你盯着各个部门的回复，谁的回复到了，同事就叫你去处理。你去处理 A 部门回复的时候，B 部门的回复可能还没到，但你不用管，继续处理 A。处理完 A，你回到工位，同事又告诉你 B 的回复也到了。整个过程中，你这个人没有同时做两件事，但你的时间被高效地压榨了。
这也就是为什么 asyncio 适合 IO 密集型任务的本质原因。IO 等待时间在总时间中占比越高，事件循环带来的收益就越大。如果任务是计算密集型的，比如一个纯数学运算的循环，那事件循环反而会因为需要反复 check、反复调度而增加额外开销。
2.3 一个任务的生命周期：从创建到回收
我总结一下一个 Task 从出生到结束经历的完整阶段，这对后续排查问题非常有帮助：
- 创建：用
asyncio.create_task(coro)或者loop.create_task(coro)把协程对象包装成 Task 对象，此时协程并没有马上执行，只是被安排进了事件循环的就绪阶段。 - 首次调度：事件循环在下一个 tick 会取到这个 Task，开始执行协程代码，一直运行到第一个 await 挂起点。
- 等待挂起：Task 在 await 一个 Future/sleep/IO 的时候会被挂起，事件循环暂时不管它。
- 唤醒：底层 IO 事件准备好、或 sleep 时间到、或 Future 被外部设定结果时，事件循环把这个 Task 重新放入就绪队列。
- 继续执行：Task 从上次暂停的 await 语句处恢复，继续往后执行。
- 完成：协程函数 return，Task 的 result 被设置，缓存在 Task 对象里。
- 回收：Task 对象如果没有其他引用，会被垃圾回收。如果它执行过程中抛出了异常且没人接收，事件循环会打印一条警告，但默认不会导致程序崩溃。
这个生命周期中值得多看一眼的是第 5 步，await 恢复之后，它是从调用 await 的地方的下一行继续执行的。很多人以为 await 是把整个函数都挂起，其实不是，它只是暂停当前协程，但是暂停点之前的局部变量、控制流状态都被保存着，恢复后原封不动接着往下跑。这就是 Python 协程与线程最显著的差异之一：线程是被操作系统一视同仁地调度，协程是自发的、有条不紊的协作。
3. 代码层面看懂事件循环的调度顺序
3.1 用事件循环 API 管理程序生命周期
写 asyncio 代码，你首先接触的可能是这三兄弟：asyncio.run()、loop.run_until_complete()、loop.run_forever()。很多初学者把这些 API 混在一起用，遇到复杂组合就晕。我先说下我的理解。
asyncio.run()是 Python 3.7 引入的，现在写简单脚本最推荐的入口。它的本质是帮你完成了三件事：内部创建新的事件循环、运行传入的协程直到它完成、最后关闭事件循环。你可以理解为它是给你专门开了一个干净的房间（事件循环），在里面干完活，再把房间锁起来、清掉。所以你会发现这个函数定义里有一个细节：asyncio.run()每次都会创建一个全新的独立事件循环，并且要求当前线程不能有正在运行的事件循环，否则会报 RuntimeError。
如果你用旧代码，可能会看到loop.run_until_complete(coro)这种写法，它表示在现有事件循环对象上，只运行某个协程直到它完成。区别在于run_until_complete不会自己关闭事件循环。如果你在同一个事件循环上第二次调用run_until_complete，是可以的，只要事件循环没关闭；但如果你直接创建一个新的循环对象，旧的又没关，在调试模式下会有资源泄漏提示。
run_forever()则完全是另一回事，它会让事件循环永远转下去，直到内部有人调用了loop.stop()。它可以理解为：事件循环进入一个“永不落幕的调度大厅”，各种任务来了就走、走了又来，谁都不负责叫停。
平时我个人推荐，写一次性脚本优先asyncio.run()；如果你的程序是常驻进程（比如一个用 asyncio 实现的 socket 服务器），那就用run_forever()，或者更现代的做法是用asyncio.start_server创建完服务器后，往asyncio.run(server.serve_forever())这种模式靠。我这里顺便说一句，服务端场景下你最好把loop对象的生命周期掌控得很清楚，因为它牵扯到信号处理、连接清理等一系列问题，后面会展开说。
3.2 演示：用不同方式创建任务，看事件循环的调度顺序
先来一段最简单的代码，看懂控制台输出顺序，你就基本懂了调度：
import asyncio async def say_after(delay, msg): await asyncio.sleep(delay) print(msg) async def main(): print("开始") task1 = asyncio.create_task(say_after(1, "1秒后执行")) task2 = asyncio.create_task(say_after(2, "2秒后执行")) print("任务已创建，等待执行") await task1 await task2 print("结束") asyncio.run(main())控制台输出顺序为：
开始 任务已创建，等待执行 1秒后执行 2秒后执行 结束你有没有发现，两个任务总共只花了 2 秒而不是 3 秒？这就是事件循环并发调度的结果。两个say_after协程同时进入了事件循环，它们都运行到了await asyncio.sleep(...)，都把自己挂起了。事件循环用 call_later 分别注册了 1 秒后和 2 秒后的定时唤醒。1 秒没到的时候，主函数main它自己也已经在await task1处挂起了，所以循环里没有任何可执行的就绪任务——事件循环在这一秒内是真正地“闲置”在系统调用上的，不是空转。1 秒后 task1 醒，打印；2 秒后 task2 醒，打印。
如果你能把这里的每一步状态转换画出来，事件循环就理解了一半。我再给个建议：初学阶段在 Python 3.11 之后的版本，打开 asyncio 的调试模式，配置PYTHONASYNCIODEBUG=1环境变量或者调用loop.set_debug(True)，然后在代码里观察调度的次数，会帮助很大。不过默认环境可能没有这个提示，这里不做展开。
3.3 事件循环不只是“执行协程”，还会执行“回调”
现在很多介绍 asyncio 的文章，通篇讲的都是async/await，导致读者误解事件循环只有一个类型的工作要处理，就是协程。但事件循环还有一套独立的机制，就是回调。我当初学 asyncio 时，是把“协程调度”和“回调机制”分开理解的，两者的语法差异很大，但对底层来说都只是“到某个时刻把一段代码放到就绪队列里去执行”。
看这段代码：
import asyncio def my_callback(name): print(f"回调 {name} 执行") async def main(): loop = asyncio.get_running_loop() # 立即放入下一轮就绪队列 loop.call_soon(my_callback, "soon") # 0.2 秒后执行 loop.call_later(0.2, my_callback, "later") print("协程主逻辑开始") await asyncio.sleep(0.1) print("协程主逻辑结束") asyncio.run(main())这里的执行顺序是：先打印“协程主逻辑开始”，然后事件循环在 0.1 秒后把主协程唤醒，打印“协程主逻辑结束”，然后紧接着执行call_soon注册的 “soon” 回调，再执行 0.2 秒后到期的 “later” 回调。这里可能需要补充一下背景，因为main在await asyncio.sleep(0.1)时，它打印完“协程主逻辑开始”之后就把控制权还给事件循环了，事件循环发现当前没有可执行的就绪任务，就进入休眠等待。到 0.1 秒main醒了，继续打印“协程主逻辑结束”，接着结束。在这个时间线上，call_soon注册的“soon”是在第一个可执行的机会执行的，但事件循环在注册时正处于 main 协程还没让出的状态中，所以它要等 main 协程下次让出控制权后才能执行。
这个细节很多人会弄错。call_soon的回调并不是马上执行，而是在“当前正在运行的任务主动让出控制权之后”被执行。也就是说，回调注册只是把它排到了就绪队列的队尾。如果当前协程一直在运行、一直不让出，那回调就永远没有机会执行。
这也解释了为什么“在事件循环内部不要写同步阻塞调用（如 time.sleep）”是铁律，因为time.sleep会让整个线程休眠，事件循环根本没有机会去调度回调和其他任务。如果你在 async 代码里要模拟一个“轮询”类操作，请不要直接用time.sleep阻塞，可以用await asyncio.sleep。这两个的区别，一个把整个线程都睡死，一个只把当前任务挂起并让事件循环去执行别的任务。
3.4 事件循环的调试思维：如何“手动”驱动一个循环
有个非常古朴的调试技巧：用loop.call_soon来观察每一轮循环的先后顺序。如果你写一段没有任何 IO、纯由回调触发的代码：
import asyncio async def main(): loop = asyncio.get_running_loop() for i in range(5): loop.call_soon(lambda i=i: print("回调", i)) await asyncio.sleep(0) asyncio.run(main())因为main里没有其他让出点，await asyncio.sleep(0)是关键一行，它把控制权让回给事件循环。事件循环这时才会依次把 5 个回调都执行完，再回到 main。如果去掉那个await asyncio.sleep(0)，这 5 个回调会一直排队，直到 main 协程结束、整个asyncio.run退出事件循环时才被清理，你在输出里根本看不到它们。
所以我会特别强调一个结论：事件循环只在 await 处才能发生任务切换。你要想让多个任务并发，必须保证每个任务里面都含有 await 让出点。如果某个协程一执行就是一个几十亿次的 for 循环、从不让出，那它就会霸占整个事件循环，其他协程全部饿死。这也是为什么异步爬虫框架里每个 HTTP 请求都会走类似await session.get(url)这样的代码，因为网络库内部本质就是创建 socket 并注册进事件循环，然后自己挂起。
4. 事件循环的创建、嵌套与线程模型
4.1 一个线程里可以有几个事件循环
Python 官方文档明确说，在同一个线程内，任意时刻最多只能有一个运行中的事件循环。这个设计是为了避免调度歧义：如果同一线程有两个循环同时在运行，就绪队列和定时器队列都会乱套。
但这里有一个经常让人困惑的边界：同一线程里，可以有多个事件循环，但不能同时“运行”。比如你可以先创建一个 loop1，运行完run_until_complete关闭它，然后再次new_event_loop创建 loop2。这种顺序创建的模式是可以的。需要注意，如果没有显式关闭 loop1，某些操作系统资源可能得不到释放，久而久之会有文件描述符泄漏。
再常见的场景是线程和事件循环的搭配。每个线程都可以创建并运行自己的事件循环，它们互不干扰。比如你有一个主线程专门跑 UI 事件循环，还有一个后台线程跑 asyncio 事件循环来处理网络请求，这在架构上是允许的。前提是：千万不要跨线程直接给另一个线程的事件循环投递任务时不做任何加锁或调用线程安全方法。
如果你确有必要从另一个线程往某个事件循环里丢任务，官方提供了loop.call_soon_threadsafe。它的内部实现会做必要的线程间同步，并唤醒目标事件循环所在的线程（通过创建一个管道或类似机制），让目标循环立即处理新来的回调。注意，Task 对象本身不是线程安全的，普通的create_task也不能从其他线程直接调用，需要配合call_soon_threadsafe使用。
4.2 asyncio.run 与嵌套事件循环的“沙箱陷阱”
这个坑我已经踩过不止一次了，写出来给各位避雷：在你已经运行着一个事件循环的时候，如果代码里再调用asyncio.run()，会抛RuntimeError: asyncio.run() cannot be called from a running event loop。这种报错在下面的场景中极其常见：
- 你在一个 async 函数里调用了另一个用
asyncio.run来写的库函数，比如有人把封装好的抓网页函数写成def get_html(): return asyncio.run(fetch())，结果你在async def main里去调了它。 - 在 Jupyter Notebook 里，因为 notebook 的内核已经运行了一个事件循环，你在 cell 里写
asyncio.run(...)，大概率会报同样的错。
关于这种嵌套场景，有几种常规的处理办法。最简单的，如果你的代码中出现了这种结构，我建议你重构：要么不要在一个已经运行的 loop 里启动新的循环，要么用asyncio.create_task在现有事件循环上调度新的协程。在 Jupyter 场景中，可以用await语法直接让 notebook 的内核事件循环来调度你的异步代码。
另一种场景，“我确实需要一个新事件循环，又不想污染当前环境的循环”，很多库会使用asyncio.new_event_loop()创建独立循环，然后在该循环里跑一些隔离逻辑，最后显式 close。这种做法我遇到过坑：如果循环里有尚未完成的循环任务，它会警告你不应该关闭有 pending 任务的循环。所以我建议除非确有必要，否则尽量别开新的事件循环，共享同一个运行中的循环是更安全的选择。
4.3 Windows 平台上的事件循环细节
我平时主要用的是 Linux 和 macOS，但给一些同事调试 Windows 下的 asyncio 程序时，也见过不少平台相关的问题。在 Windows 上，Python 3.8 之前默认的事件循环是SelectorEventLoop，底层用的是 select，它在 Windows 上有很多别扭的限制，比如不能监听管道、某些 socket 事件会失效。从 Python 3.8 开始，asyncio 在 Windows 上默认使用ProactorEventLoop，底层利用 IOCP，能更好地支持异步 socket 等特性。
具体到一个很影响开发的细节：Windows 上asyncio对子进程的支持非常依赖ProactorEventLoop，如果你手动强制把 Windows 的事件循环改成SelectorEventLoop，那asyncio.create_subprocess_exec等子进程相关功能会直接不可用或报错。所以如果你发现 Windows 上子进程相关的异步代码行为诡异，先检查一下运行时当前事件循环的类型。
再补充一个，Windows 控制台对信号量的支持极其有限，比如loop.add_signal_handler在 Windows 上是不受支持的，因为信号机制本身就不是 Windows 的一等公民。如果跨平台程序里要用到 Ctrl+C 优雅退出，需要单独处理，而不能照搬 Linux 的写法。
4.4 事件循环与 Future/Task 的绑定关系
还有一个隐藏较深的知识点，很多人不知道，就是 Future 和 Task 在创建时就会绑定一个事件循环，通常就是当前正在运行的那个。当这个 Future 完成时，它会通过内部持有的 loop 引用来安排回调的执行。所以如果你把一个 Future 从线程 A 的事件循环里拿给了线程 B 的异步代码用，指望线程 B 的 await 能响应它，那就会出问题。跨事件循环传递 Future、Task，本质上就是错误，因为这些对象与它们的“家”是绑定死的。
我自己就碰到过“await 永远不会返回”这种棘手问题。最后排查发现原因就是：代码里用了concurrent.futures.Future和 asyncio 的 Future 混用，loop.run_in_executor返回的一直是 asyncio 的 Future，外部却用线程池的 Future 去等它，结果两边没对上。要解决这种跨环通信，正统做法是使用asyncio.wrap_future包裹一个 concurrent Future，或者反过来用asyncio.futures.wrap_future转换，使得它们能被当前事件循环所识别。
5. 事件循环编程中的常见问题与排查技巧
5.1 一碰就炸的“Event loop is closed”
这个错误我见得太多次了，通常发生在你试图在一个已经关闭的事件循环上创建任务或运行协程时。它最常见的触发场景，是在一个用asyncio.run写好的函数退出之后，外部代码还引用着这个函数内部的 loop 对象或者 Task 对象，想要再靠它执行点什么。
比如下面这种代码：
import asyncio async def inner(): print("inner") async def main(): task = asyncio.create_task(inner()) await task return task task = asyncio.run(main()) # 此时 main 内部的事件循环已经关闭 try: task.get_loop() except RuntimeError as e: print(e)解决方案不是去“重新打开”一个已经关闭的循环，而是设计上要避免跨循环持有这些对象。老一点的代码喜欢用loop = asyncio.get_event_loop()全局获取循环，然后到处用这个 loop，这类写法在 Python 3.10 之后也会逐步被淘汰，因为你不知道它拿到的循环是否还能用。
所以，我认为，现在写 asyncio 代码最安全的姿势就是：不要显式保存或到处传递 loop 对象，如果确实需要 event loop，在协程内部调用asyncio.get_running_loop()动态获取。get_event_loop这种老 API，在已经运行的事件循环外部调用时容易摸不清状态。
5.2 NotImplementedError：没有正在运行的事件循环？
还有一个非常常见的报错，RuntimeError: no running event loop，它和上面那个是难兄难弟。原因是调用某个 asyncio 接口时，当前线程里没有正在运行的循环。最常见的位置是asyncio.get_event_loop()、asyncio.new_event_loop()被误用，以及你写了一个函数，它没有 async 类型却想直接调用asyncio.sleep，这属于语法层面的错误，因为await只能在 async 函数里用。
但是有一个情况很多人会陷进去：在一个同步函数里创建 Task。实际上asyncio.create_task在 3.10 之后必须在一个正在运行的事件循环内调用。如果你写一个普通的def func()，然后在里面调用asyncio.create_task(...)，大概率直接报RuntimeError: no running event loop。
正确的做法是，同步函数里不要直接创建 Task。要么把同步函数改成 async 函数，要么在同步函数里先启动一个循环并运行目标协程。如果你是需要把 Task 抛到后台去，可以明确让主协程先把循环跑起来，然后用loop.call_soon_threadsafe等方式投递。
5.3 Task was destroyed but it is pending
这个警告通常意味着你有协程任务，创建了但没有等待它完成，程序就退出了。事件循环在关闭或者清理的时候，发现有 Task 对象还在 pending 状态，于是警告你。常见场景：
import asyncio async def never_finish(): await asyncio.sleep(100) async def main(): task = asyncio.create_task(never_finish()) # 没有 await task，main 就结束了 return asyncio.run(main())这时你会收到一个警告，提醒你 Task 被销毁时还处于挂起状态。处理方式有几种：
- 如果真的不关心任务完成，可以
task.cancel()或者让它脱离主流程，但千万别让它泄漏。 - 如果任务需要继续跑直到结束，确保
main函数里await task或使用await asyncio.gather(task)。 - 如果想实现“丢到后台跑”，考虑用
asyncio.create_task创建后，把 task 持有在一个集合里，同时注册一个 done 回调用于清理。
我给出的一个比较实用的模式是这样的：
import asyncio pending_tasks = set() async def background_work(): for i in range(5): await asyncio.sleep(1) print("后台工作", i) async def main(): task = asyncio.create_task(background_work()) pending_tasks.add(task) task.add_done_callback(pending_tasks.discard) print("main 不等待后台任务") await asyncio.sleep(2) asyncio.run(main())这样 main 只跑 2 秒，后台任务却一直会继续到第 4 秒，事件循环会在所有 pending 任务都结束之后才退出吗？答案是不会！这里就是另一个坑：asyncio.run(main())只有当 main 协程结束时就会关闭事件循环，不会主动等待后台任务。所以如果你想让后台任务继续执行，应该在 main 内部也await task，或者确保调度任务的事件循环不是在 main 结束时立刻退出。
5.4 阻塞事件循环的隐形杀手：同步 IO 和 CPU 密集计算
这个坑非常隐蔽，几乎所有初学 asyncio 的人都掉进去过。你以为用 async 定义了函数就是异步了，然后函数体里面居然用的是requests.get(...)，这是个同步的 HTTP 请求库。当第一个任务执行到resp = requests.get(url)时，整个事件循环所在的线程就被阻塞住了，其他任务全部冻结。你能看到的现象是：程序串行执行了，一个请求等完再发下一个，耗时爆炸。这种 bug 不会直接报错，非常折磨人。
所以每当我看到有人在 async 函数里写同步 IO 库，我就会直接提醒他：换异步库。比如requests换成httpx.AsyncClient或aiohttp；文件操作要避免同步做，可以考虑asyncio.to_thread扔到线程池里；数据库查询如果是同步 ORM，要慎重考虑。
如果不是 IO 阻塞而是 CPU 密集计算，e.g. 一个解析超大 JSON 的函数，那就用loop.run_in_executor(None, fn)把它丢到线程池甚至进程池中去执行，不然它会把事件循环卡死。Python 的 asyncio 是一个绝佳的外部 IO 调度器，但它在“内部计算”上的并行能力约等于零。想要 CPU 密集并行，需要配合多进程，让多个进程各自运行独立事件循环，才能做到真正的并行利用多核。
5.5 排查工具不到位？分享几个调试思路
真正排查 asyncio 问题的时候，光靠print是不够的。我常用的有三个方法：
一是打开 asyncio 调试模式，PYTHONASYNCIODEBUG=1或代码里设置loop.set_debug(True)。开启后，事件循环会检测哪些协程执行时间过长、会检查是否有回调执行超时，甚至会输出 future/task 的创建堆栈，定位未完成任务来源特别有用。
二是使用asyncio.all_tasks(loop=loop)来查看当前事件循环中还有哪些 Task 在跑。你可以为每个任务打印它的协程来源、状态、创建它的堆栈。这个方法在排查“程序退出时卡住”或者“为什么 Task 还没结束”时非常直接。
三是记清楚协程和 Task 的状态变化时间线。我在排查具体问题时，习惯写一个简单的统一日志装饰器，把任务开始、await 某个地方、唤醒、结束这四个点都打日志。这样虽然代码丑一点，但对理清调度顺序是肉眼可见的直观。
6. 更多关于事件循环的经验总结
6.1 我对事件循环心智模型的最终总结
如果你把事件循环理解成一个“消息泵”，其实也通：所有想被调度执行的代码都会被包装成消息（回调或 Task step），扔进泵里，泵不关心消息之间是不是同一个协程，它只负责在合适的时间逐个抽出去执行。这个泵的转速以及什么样的时间算“合适”，是由底层 IO 事件和定时器共同决定的。
我记得自己初学 asyncio 时，专门去读 CPython 的Lib/asyncio/base_events.py源码，想搞清楚_run_once里到底是怎么干的。翻了一会儿我就发现，它先检查定时器队列、计算休眠时间、调用 epoll 等待、处理 IO 事件、将对应的 future 标记为完成、然后运行就绪队列里的回调。在这些步骤里最微妙的是：IO 事件对应的 future 被标记完成之后，并不是立即切换到对应协程去执行，而是要把它的回调放到就绪队列里，等当前单次循环的 IO 事件处理完，才会去跑这些回调。理解这部分，你就不会再搞混“为什么事件循环处理事件有延迟”的问题了。
6.2 在实际项目中如何“组织”事件循环的生命周期
写一个短脚本，asyncio.run已经够用；写一个服务器，事件循环的生命周期一般要和应用进程的生命周期一致。比如你在写一个基于 asyncio 的 WebSocket 推送服务，它的主协程通常是：
import asyncio async def main(): server = await asyncio.start_server(handle_client, "127.0.0.1", 8888) async with server: await server.serve_forever() async def handle_client(reader, writer): while data := await reader.readline(): writer.write(data) await writer.drain() writer.close() await writer.wait_closed()事件循环因为serve_forever()而持续运行。如果你希望它优雅退出，需要注册signal.SIGINT和signal.SIGTERM的处理函数，让它在收到信号时取消任务并关闭 server。现代写法上，你可以用asyncio.Runner这个上下文管理器来显式管理事件循环，它比裸用asyncio.run多一点灵活性。
针对asyncio.Runner，我简要说一下，Python 3.11 引入它是为了让事件循环资源的获取和释放更清晰。它的用法类似：
import asyncio async def main(): await asyncio.sleep(1) with asyncio.Runner() as runner: runner.run(main())相比asyncio.run，它允许你在同一个 Runner 内多次run()协程，而不会每次重新创建全新的循环。这对于某些需要在同一个事件循环里反复执行异步任务的测试框架、脚本环境非常有意义。
6.3 不要盲信“线程池 + asyncio”的组合
有些开发者在遇到阻塞任务时，想到的第一方案是把阻塞代码丢到asyncio.to_thread。这个函数的底层用的是默认线程池ThreadPoolExecutor，是可行的，但你得想清楚：to_thread只是把阻塞操作“搬出了事件循环”，让事件循环不被卡死，不代表程序整体并发能力变强。每个额外线程都会带来上下文切换开销，IO 密集的大规模并发（比如上万网络连接）用线程池，系统资源会急剧吃紧。事件循环的真正威力在于它能把成千上万个 IO 等待都交给内核的 epoll/IOCP 去统一管理，而不是靠开线程堆资源。所以，事件循环 + 异步 IO 库永远是最贴近异步范式的手段；线程池应该用来处理那种“无法改写成异步的遗留库”，比如某些同步数据库驱动，不能拿来模拟所有场景。
我在实际处理爬虫业务时，框架选型上也经常看到有人把asyncio.Semaphore配合线程池一起用，控制并发连接数。Semaphore这个原语本身在异步世界里也有特殊意义，它是协程间协作的资源控制器，可以在 await 时阻塞协程而不是阻塞线程，它不会破坏事件循环的调度。使用asyncio.Semaphore时，你可别把它与threading.Semaphore混着用，后者会阻塞事件循环，前者不会，两者底层完全不同。
6.4 关于高版本 Python 中事件循环的发展
我近几年在关注 Python 版本升级，一个重要变化是 Python 3.10 起asyncio.get_event_loop在不存在运行循环时的行为发生了变化：如果当前没有运行中的事件循环，它可能会发出 DeprecationWarning，未来版本可能直接报错。这背后是为了强制开发者明确所使用的 loop 来源，不再隐式地创建一个全局默认循环。很多老代码库却依然在顶层同步代码里调用asyncio.get_event_loop()来拿 loop，再基于它去创建任务。这些代码在 Python 3.10 之后陆续会出现警告或问题。
我个人的看法是，除非兼容老版本库，否则新代码一律用asyncio.run+await，需要 task 时在协程内部用asyncio.create_task，需要运行中的 loop 时用asyncio.get_running_loop。这套组合能让你远离百分之九十的循环生命周期管理问题。
6.5 最后的最后分享一个排查技巧
如果你写一个 asyncio 程序时，莫名其妙发现程序卡住了，怎么打日志都查不出来卡在哪一行，我会建议你开一个后台线程，定时打印当前事件循环中所有任务的状态和堆栈。具体做法是启动一个普通 daemon 线程，在里面循环执行：
import asyncio import threading import time def dump_tasks(loop): while True: time.sleep(3) tasks = asyncio.all_tasks(loop) for task in tasks: if not task.done(): print("任务可能卡住:", task) task.print_stack() async def main(): loop = asyncio.get_running_loop() threading.Thread(target=dump_tasks, args=(loop,), daemon=True).start() await asyncio.sleep(100) asyncio.run(main())这个task.print_stack()会输出该任务当前执行到的代码堆栈，能直接定位到阻塞点。我自己曾用这个技巧很快找到了一个底层同步 socket 阻塞事件循环的 bug——那个函数我怎么看都像是异步的，但一打印任务堆栈发现实际代码停在了一个同步 socket 的recv调用。所以说，工具在排查异步问题方面真的比直觉靠谱得多，这也是事件循环开发里我最后想跟大家分享的一条实战经验。
