asyncio 事件循环与协程

场景:一万个慢请求,如何用一个线程扛下来

你要写一个爬虫,抓一万个 URL,每个请求等待远端响应大约 200ms。用同步代码,一个一个抓,总耗时是 10000 × 200ms = 33 分钟。用多线程能并发,但开一万个线程,光线程栈内存就要几个 GB,操作系统调度也吃不消。

asyncio 给出的答案是:用单个线程、单个事件循环,承载上万个并发的"等待"。因为这些请求 99% 的时间都在等网络,真正占用 CPU 的只有发起请求和处理响应那一瞬。既然如此,何必为每个等待都开一个昂贵的线程?用协程把"等待"挂起,让一个线程在上万个挂起点之间快速穿梭,才是更经济的模型。

机制一:协程是"可暂停的函数"

理解 asyncio,先得理解协程(coroutine)和普通函数的根本区别。普通函数一旦调用就一口气执行到 return,中途无法暂停。协程不一样——用 async def 定义,它可以在 await 处主动暂停,把控制权交还给调用者,稍后再从暂停点恢复。

async def fetch(url):
    print("发起请求", url)
    data = await http_get(url)   # 在这里暂停,等 IO,期间让出控制权
    print("收到响应", url)        # IO 好了之后从这里恢复
    return data

关键认知:调用 fetch(url) 不会执行任何代码,它只是创建一个协程对象。协程是惰性的,必须交给事件循环去驱动(await 它,或 asyncio.create_task),它才会真正跑起来。这点和很多人的直觉不同——一个常见的 bug 就是写了 fetch(url) 却忘了 await,函数体根本没执行,还收到一个 “coroutine was never awaited” 警告。

协程的底层是生成器演化而来的。await 在机制上类似生成器的 yield:它把一个值(具体说是一个"我在等什么"的信号)抛给上层的驱动者,并保存当前所有局部变量和执行位置,等被恢复时原样继续。

机制二:事件循环如何调度

事件循环(event loop)是 asyncio 的心脏。它本质是一个单线程的无限循环,做三件事:运行就绪的任务、监听 IO 事件、在合适的时机恢复被挂起的协程。

简化的循环逻辑大致是:

while 还有任务或事件:
    1. 跑所有"立即就绪"的回调(ready 队列)
    2. 计算最近的定时器到期时间,作为 selector 的超时
    3. selector.select(timeout)  # 用 epoll/kqueue 监听所有注册的 fd
    4. 哪些 fd 就绪了(socket 可读/可写),就把对应协程的恢复回调放进 ready 队列
    5. 到期的定时器(如 asyncio.sleep)也放进 ready 队列
    回到第 1 步

当一个协程 await 一个网络读时,它不会阻塞线程,而是:把自己的 socket 注册到事件循环的 selector 上,然后挂起、交出控制权。事件循环转去跑别的就绪协程。等到 selector 报告这个 socket 可读了,事件循环再把这个协程恢复,从 await 之后继续。

单线程,但永远在干活:
时刻t1: 协程A await socket读 → 挂起,注册fd → 循环转去跑协程B
时刻t2: 协程B await socket读 → 挂起 → 循环转去跑协程C
时刻t3: A的socket就绪 → 恢复A,处理数据
...一个线程在成千上万个挂起的协程之间高速穿梭,没有一刻空等

这就是单线程能扛上万并发连接的奥秘:没有线程在傻等,所有"等待"都被转化成了 selector 上的一个 fd 注册。

源码视角:并发的正确姿势

新手最容易犯的错是串行地 await,以为加了 async 就并发了:

# 错误:这是顺序执行,总耗时 = 所有请求之和
async def fetch_all_wrong(urls):
    results = []
    for url in urls:
        results.append(await fetch(url))   # 一个等完才发下一个
    return results

每个 await fetch(url) 都会等当前请求彻底完成才进入下一轮循环,根本没并发。要真正并发,得先把协程都变成**任务(Task)**丢进事件循环,再一起等:

# 正确:所有请求并发发出,总耗时 ≈ 最慢的那个
async def fetch_all(urls):
    tasks = [asyncio.create_task(fetch(url)) for url in urls]
    return await asyncio.gather(*tasks)

create_task 把协程立即调度进事件循环开始执行,gather 则等待这一批全部完成。差别在于:前者所有请求的"等待 IO"是重叠的,后者是串行的。理解这一点,是用对 asyncio 的分水岭。

现代写法更推荐 asyncio.TaskGroup(结构化并发),它能在任一子任务出错时自动取消其余任务,避免任务泄漏。

工程权衡与踩坑

铁律一:绝不能在事件循环里调用阻塞操作。 asyncio 是单线程的,一旦你在某个协程里调了同步的 requests.get、time.sleep、或一个 CPU 密集的死循环,整个事件循环就被卡死——所有其它协程全部停摆,因为没人来跑那个循环了。这是 asyncio 最致命的坑。需要做阻塞/CPU 工作时,要丢到线程池或进程池:

1 2	# 把阻塞调用扔到线程池,不堵塞事件循环 result = await asyncio.to_thread(blocking_db_query, sql)

铁律二:生态必须全是异步的。 asyncio 的优势建立在"所有 IO 库都用非阻塞方式"之上。你必须用 aiohttp 而不是 requests,用异步数据库驱动而不是同步的。一旦链路里混进一个阻塞库,前面说的"卡死整个循环"就会发生。这意味着迁移成本和库选型限制是实打实的。

误区:以为 asyncio 比多线程"更快"。 对单个任务的执行速度,它并不更快——它的优势是用极低的内存代价支撑极高的并发数。协程的开销远小于线程(没有线程栈、没有内核态调度),所以同样的内存能开的协程数是线程的几个数量级。但对 CPU 密集任务,asyncio 毫无帮助,反而因单线程特性更糟,那是多进程的领域。

还有一个隐蔽坑:异常被吞。 如果一个用 create_task 创建的任务出错了,而你从没 await 过它,异常不会立刻抛出,只会在任务被垃圾回收时打印一条警告,很容易被忽略。务必妥善 await 或用 TaskGroup 管理每个任务。

边界:什么时候该用 asyncio

适合:海量网络 IO 并发(爬虫、API 网关、聊天/推送服务、大量下游扇出调用)、长连接(WebSocket)。

不适合:CPU 密集计算(用多进程)、依赖大量成熟同步库且无异步替代的项目(改造成本可能高于收益)、逻辑简单的低并发脚本(同步代码更易读易调)。

小结

asyncio 的核心是一个单线程事件循环,配合可在 await 处暂停恢复的协程,把每一次"等待 IO"都转化成 selector 上的一个 fd 注册,从而让一个线程在上万个挂起的协程之间高速穿梭、没有一刻空等。用对它的关键有两条:一是真正让请求并发(用 create_task / gather / TaskGroup,而非串行 await),二是死守"事件循环里不能阻塞"的铁律、全链路使用异步库。它换来的不是单任务更快,而是用极低内存支撑极高并发的能力——认清这个定位,你才不会拿它去硬扛 CPU 密集任务,也不会再写出那种加了 async 却依然串行的"假并发"代码。