Python asyncio：事件循环不是魔法

大多数 asyncio 教程都在讲 API，很少有人讲清楚它到底在解决什么问题，以及它在哪里会让你翻车。

这篇文章不是文档搬运。它来自真实项目里踩过的坑。

---

异步的本质是什么

先把一个误解排掉：async/await 不让你的代码跑得更快，它让你的 CPU 在等 I/O 的时候不闲着。

这个区别很关键。如果你的瓶颈是 CPU 密集计算（矩阵乘法、图像处理、模型推理），asyncio 帮不了你，你需要的是多进程或者 GPU。asyncio 的价值域是 I/O 密集型任务：网络请求、数据库查询、文件读写。

理解这一点之后，再去看事件循环就会清晰很多。

---

事件循环：一个不断轮询的调度器

事件循环的工作模式本质上是：

while True:
    events = selector.select(timeout)  # 向 OS 询问：有哪些 fd 可读/可写了？
    for event in events:
        callback = event_map[event.fd]
        callback()                      # 执行对应的回调
    run_ready_callbacks()               # 执行当前轮次积累的即时回调

selector.select() 背后是操作系统的 I/O 多路复用（epoll/kqueue/select），它是系统调用，阻塞等待直到有 fd 就绪或超时。asyncio 的”异步”能力完全来自这里，而不是来自什么 Python 层的魔法。

这就是为什么 在协程里做任何阻塞调用都会挂死整个事件循环。time.sleep(1) 不会挂起当前协程，它会挂起整个线程，而整个线程只有一个事件循环。

# 错的
async def fetch_data():
    time.sleep(1)       # 阻塞了事件循环，其他所有协程都得等
    return result

# 对的
async def fetch_data():
    await asyncio.sleep(1)  # 挂起这个协程，事件循环继续跑其他任务
    return result

常见的阻塞陷阱：

• requests.get() → 用 aiohttp 或 httpx
• time.sleep() → 用 asyncio.sleep()
• 同步的 DB 客户端 → 用对应的 async driver
• subprocess.run() → 用 asyncio.create_subprocess_exec()

---

协程的本质：可暂停的函数

async def 定义的不是普通函数，是协程函数。调用它不会执行代码，只会返回一个协程对象：

async def say_hello():
    print("hello")

coro = say_hello()   # 什么都没打印
await coro           # 现在才执行

协程的暂停点是 await。当协程执行到 await some_io_op() 时，它把控制权交还给事件循环，事件循环可以去跑别的协程，等 I/O 完成后再恢复这个协程。

这个机制来自 Python 的生成器协议。async def 是 yield 的语法糖升级版——协程是有状态的可暂停函数，它的局部变量在暂停期间保持完整。这也是它比线程轻量的原因：切换协程不需要内核参与，只是 Python 层的函数调用。

---

Task 和 Future：并发的基础单元

协程是串行执行的，想要并发得靠 Task：

# 串行：total time = 2s + 3s = 5s
async def main():
    r1 = await fetch("url1")   # 等 2s
    r2 = await fetch("url2")   # 等 3s

# 并发：total time ≈ max(2s, 3s) = 3s
async def main():
    t1 = asyncio.create_task(fetch("url1"))
    t2 = asyncio.create_task(fetch("url2"))
    r1 = await t1
    r2 = await t2

create_task() 把协程包装进 Task 并立即调度，不等 await。gather() 是更常用的并发原语：

results = await asyncio.gather(fetch("url1"), fetch("url2"), fetch("url3"))

Future 是更底层的概念——一个占位符，代表”还没得到的值”。Task 是 Future 的子类。大多数情况下你不需要直接操作 Future，除非你在写底层的异步适配器（比如把回调式 API 包装成 async 接口）。

---

三个在生产里最容易翻车的地方

1. gather 的异常传播

results = await asyncio.gather(task1(), task2(), task3())

如果 task2 抛出异常，默认情况下 gather 会把异常抛给调用方，但 task1 和 task3 会继续跑——它们成了无人认领的后台任务，可能在任意时刻抛出 Task exception was never retrieved 警告。

正确做法：

# 方案 A：return_exceptions=True，把异常作为返回值处理
results = await asyncio.gather(task1(), task2(), task3(), return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        handle_error(r)

# 方案 B：Python 3.11+ 的 TaskGroup，任意一个失败则取消其余
async with asyncio.TaskGroup() as tg:
    t1 = tg.create_task(task1())
    t2 = tg.create_task(task2())

TaskGroup 是 Python 3.11 引入的”结构化并发”实现，它保证了：子任务的生命周期不会逃逸出 async with 块。这和 Go 的 errgroup 是同一个思路。

2. 取消不干净

task.cancel() 只是在协程下一个 await 点注入一个 CancelledError，不保证立刻停止。协程可以捕获它，做清理，然后必须重新 raise：

async def worker():
    try:
        while True:
            await do_work()
    except asyncio.CancelledError:
        await cleanup()    # 清理资源
        raise              # 必须重新抛出，否则取消被吞掉

如果协程里有 try/except Exception，它会把 CancelledError 也吞掉（CancelledError 在 3.8 之前是 Exception 的子类，3.8 之后改为 BaseException 的子类，就是为了解决这个问题）。

3. 在同步代码里调用协程

从同步代码调用异步函数：

# Python 3.7+
asyncio.run(main())

# 在已有事件循环里提交任务（跨线程场景）
future = asyncio.run_coroutine_threadsafe(coro(), loop)
result = future.result(timeout=5)

常见错误：在已运行的事件循环里调用 asyncio.run()，或者在协程里调用 loop.run_until_complete()，都会导致 RuntimeError: This event loop is already running。

---

性能：asyncio 的边界

asyncio 在 I/O 并发上表现优异，但它有天生的限制：

单线程。事件循环跑在一个线程里，CPU 计算无法并行。如果需要 CPU 密集任务，用 loop.run_in_executor() 把它扔到线程池或进程池：

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_function, arg)

GIL。即使用线程池，Python 的 GIL 也限制了纯 Python 代码的并行度。计算密集任务要真正并行必须用进程池或者 C 扩展（numpy、torch 之类会释放 GIL）。

uvloop。如果你在 Linux 上跑高并发服务，可以把事件循环替换成 uvloop——它用 Cython 基于 libuv 实现，在某些场景下吞吐量比默认实现高 2-4 倍：

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

---

选择 asyncio 之前先问自己

1. 你的瓶颈是 I/O 吗？ 如果是 CPU，asyncio 没有意义。
2. 你的代码里有没有同步阻塞？ 一个同步调用就能毁掉整个事件循环的响应性。
3. 你的团队熟悉异步心智模型吗？ 异步代码的调试比同步难，stack trace 不连续，错误传播路径反直觉。

asyncio 是好工具，但不是银弹。在一个以数据库查询和 HTTP 请求为主的服务里，它能把并发能力提升一个数量级；在一个 CPU 密集的批处理任务里，它引入的复杂度是纯粹的成本。

---

小结

• 事件循环 = OS 级 I/O 多路复用 + Python 层调度器
• 协程 = 可暂停函数，切换无内核开销
• Task = 协程的并发载体，立即调度
• 阻塞调用 = 事件循环杀手，必须清零
• 异常处理和取消传播是异步代码最容易出问题的两个地方
• Python 3.11 的 TaskGroup 是目前最稳健的并发组织方式