小李今天差点把电脑砸了。
他写了个异步爬虫，用了asyncio，每个函数前面都加了async，调网络请求的地方都写了await。代码看起来“很异步”。但跑起来之后，一万个请求花了将近一个小时。他打开调试工具一看，所有任务居然还是排着队一个一个执行。
“我不是写了await吗？为什么还是串行的？”
问题出在一个他完全没注意的细节上：他在协程里调用了requests.get()。
先看一个让你“以为懂了”的例子
小李最初写的代码大概是这样：
import asyncio
import time
async def task(name, seconds):
print(f"{name} 开始")
time.sleep(seconds) # 问题在这里
print(f"{name} 结束")
async def main():
tasks = [
asyncio.create_task(task("任务1", 2)),
asyncio.create_task(task("任务2", 1)),
asyncio.create_task(task("任务3", 3))
]
await asyncio.gather(*tasks)
asyncio.run(main())
他满心期待看到三个任务几乎同时开始、总共3秒左右结束。结果呢？任务1开始、等2秒、任务1结束、任务2开始、再等1秒、任务2结束……总共花了6秒。和同步执行一模一样。
小李盯着屏幕，心想：create_task不是并发调度了吗？gather不是同时等待吗？为什么还是串行？
答案就藏在time.sleep那一行。
同步阻塞：程序“傻等”的根源
要理解这个问题，先看一段最朴素的同步代码：
import time
def task(name, seconds):
print(f"{name} 开始")
time.sleep(seconds)
print(f"{name} 结束")
start = time.time()
task("任务1", 2)
task("任务2", 2)
print(f"总耗时: {time.time() - start}")
结果很直白：任务1开始，等2秒，任务1结束，任务2开始，再等2秒，任务2结束，总耗时4秒。
问题出在time.sleep。当程序执行到这一行时，CPU其实什么事都没干，就在那儿干等着时间过去。这段时间本可以用来处理其他任务，但同步代码不允许这样做——它是一条路走到黑的。
time.sleep是同步阻塞的，它会卡住当前线程。在普通的同步程序里，这没什么问题，反正只有这一条路。但在asyncio的世界里，这个“卡住当前线程”的行为会造成灾难性的后果。因为asyncio所有任务共享同一个线程，这个线程一旦被卡住，事件循环就彻底停摆了。
事件循环：一个单线程的“调度中心”
理解asyncio，关键要理解事件循环。
你可以把事件循环想象成一个调度中心。它维护着两个队列：就绪队列里放着可以立即执行的任务，等待队列里放着正在等待某个事件（比如网络响应、定时器到期）的任务。每一次循环，事件循环从就绪队列取出一个任务执行。当任务执行到await时，它会把自己挂起，告诉事件循环：“我先去忙别的，好了叫我”，然后事件循环就会切换到下一个可以执行的任务。
这里有一个关键认知：事件循环是单线程运行的。它不像多线程那样有操作系统帮你切换，所有的切换都由协程自己在await点主动让出。协程只在await点主动让出控制权，如果某个协程内部执行了长时间的计算而不使用await，其他协程只能干等。
这就是协作式调度——需要每个协程“配合”，才能让事件循环正常工作。而time.sleep恰恰是最不配合的那种操作：它不交出控制权，死占着线程不放。
await到底在干什么？
很多人以为await的意思是“等待”。这个理解不算错，但容易误导。
await的本质不是“等待”，而是“让出控制权”。当前协程在此处挂起，告诉事件循环“你可以去运行其他协程了，等我等的这个操作完成了再唤醒我”。
当协程执行到await时，它做了两件事：把当前的执行状态（局部变量、指令指针）保存下来，然后把控制权交还给事件循环。事件循环转而调度另一个就绪的协程执行。当被等待的操作完成时，事件循环再把之前的协程恢复执行。
所以await asyncio.sleep(1)和time.sleep(1)的区别，表面上看都是“等1秒”，但行为完全不同。
await asyncio.sleep(1)做的是：告诉事件循环“我1秒后需要被唤醒”，然后把控制权交出去。事件循环立刻去执行其他任务。1秒后事件循环回来唤醒这个协程，继续执行。
time.sleep(1)做的是：死占着线程1秒。这1秒内，事件循环完全无法运行，所有其他任务都被冻结。1秒后线程恢复，继续往下走——但中间这1秒，整个程序等于暂停了。
用一个生活场景来理解。你点了三份外卖。同步的做法是：站在第一家店门口等，拿到第一份，再去第二家等。第二家店如果忙，你就干等着，全程啥也干不了。异步的做法是：三家店都下单，然后回家坐着，谁做好了给你打电话，你去拿，中间你可以看电视、打游戏。
await asyncio.sleep就是“回家坐着”，time.sleep就是“站在店门口傻等”。
真正的坑：在异步函数里调用同步阻塞
理解了上面的原理，小李的bug就清楚了。
他用了asyncio.create_task创建了三个任务，用asyncio.gather等待它们完成。这部分写得没问题。问题在于每个任务内部用了time.sleep而不是await asyncio.sleep。
当事件循环调度到任务1时，执行到time.sleep(2)，整个线程被冻结2秒。这2秒里，任务2和任务3虽然在就绪队列里，但事件循环根本没有机会去调度它们。所以效果和同步执行一模一样。
更隐蔽的情况是调用同步的网络库。比如在协程里调用requests.get()：
async def fetch(url):
response = requests.get(url) # 同步阻塞调用
return response.text
requests.get()是同步的，它会阻塞事件循环，直到网络响应返回。这意味着即使你有100个协程同时运行，只要其中一个执行了requests.get()，其他99个都会被冻结。一个阻塞调用（哪怕只有一个）就可以冻结整个应用。
这个坑之所以“坑”，是因为代码看起来完全正确。你写了async def，你调用了await，你的代码编辑器也没有报错。但运行时，所有并发都消失了。
怎么判断自己掉坑了？
有个很简单的判断方法：看你的协程里有没有用到不包含await关键字的阻塞操作。
常见的同步阻塞操作包括：
time.sleep()— 应改为await asyncio.sleep()requests.get/post()— 应改为aiohttp或httpx的异步版本open()读取大文件 — 应用asyncio.to_thread()包装urllib相关调用 — 应改为异步HTTP客户端- 大量CPU计算（比如循环一亿次）— 应提交给进程池
open调用执行阻塞的磁盘I/O，应该在执行器中运行。与之相关的是，阻塞的读写操作也必须一并修复，否则即使修复了open，后续的读写仍会阻塞。
Home Assistant的开发文档里有一句话很直白：如果在事件循环中发生阻塞操作，在操作完成之前，其他任何东西都无法运行。因此，事件循环中不应发生任何阻塞操作，否则整个系统会在阻塞操作期间停滞。
那我不小心用了同步库怎么办？
实际开发中，你不可能把所有依赖都换成异步版本。有些库就是没有异步实现，有些第三方SDK就是同步的。这种情况下，asyncio.to_thread()就是你的救命稻草。
import asyncio
async def main():
result = await asyncio.to_thread(blocking_function, arg1, arg2)
print(result)
asyncio.to_thread是Python 3.9新增的功能，一行代码就能把同步函数变成可await的协程。它把同步函数丢到线程池里执行，不阻塞事件循环，等执行完了再把结果取回来。
但要注意一个细节：asyncio.to_thread并不是万能的。asyncio.wait_for()或asyncio.timeout()只能取消等待的协程，永远无法取消正在运行的工作线程。如果一个阻塞的系统调用（比如文件锁）卡住了，即使超时了，那个线程仍然在后台运行。
所以to_thread适合“确定能完成”的阻塞操作，不适合“可能永远卡住”的操作。后者需要用进程池或者加额外的超时和恢复机制。
更根本的问题：异步的“传染性”
还有一个很多人没意识到的问题：async/await语法具有“传染性”。
一个异步函数只能被另一个异步函数await调用。这意味着如果你有一个同步的main函数，调用了某个异步库，你需要从底到顶全链路异步化。
这不是设计缺陷，而是协作式调度的必然结果。事件循环要求所有可能阻塞的操作都以await的形式声明出来，这样它才知道在哪里切换。如果一个同步函数悄悄阻塞了，事件循环根本不知道，也就无法调度。
所以使用asyncio时，你需要接受一个现实：要么全异步，要么在同步边界用to_thread或执行器包装。混着写是最危险的。
什么时候该用asyncio，什么时候不该用？
asyncio适合I/O密集型任务——大量时间花在等待网络响应、磁盘读写、数据库查询上。当一个任务需要等待时，让程序暂时离开，去做别的事，等那个操作准备好了再回来继续。
一个HTTP请求的基准测试可以直观说明：使用同步requests库并发请求100个URL，总耗时约等于所有请求耗时的累加；使用aiohttp的异步版本，总耗时约等于最慢那个请求的耗时。
但asyncio不适合CPU密集型任务。如果协程中执行的是大量数学运算且不涉及I/O等待，事件循环会在一个协程上阻塞直到运算完成，其他协程得不到执行机会。对于CPU密集型任务，标准方案是通过run_in_executor将计算任务提交给线程池或进程池执行，主事件循环不会被阻塞。
异步本身并不能加速计算，它优化的只是“等待”这件事。
回到小李的代码
小李最终把代码改成了这样：
import asyncio
import aiohttp
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
核心改动只有两个：把requests换成了aiohttp（异步HTTP客户端），把time.sleep换成了await asyncio.sleep。
跑完一万个请求，耗时从将近一个小时降到了十分钟左右。
小李看着终端里刷刷滚动的日志，终于明白了一件事：await和同步阻塞的区别，不在于“等不等”，而在于“等的时候，能不能让别人先跑”。
理解事件循环是单线程的，理解await是“让出控制权”而不是“占用线程等待”，理解同步阻塞操作会冻结整个事件循环——这三个认知一旦建立，asyncio的大部分坑都能提前避开。
不是用了async def就是异步，关键是看代码在等待的时候，有没有把控制权交出去。
