Python异步编程与asyncio事件循环#

1. GIL 是什么，为什么 IO 密集选协程#

• GIL（Global Interpreter Lock）：CPython 解释器的全局锁，同一时刻只有一个线程在执行 Python 字节码。它保护的是解释器内部状态（引用计数等），不是你的业务数据——x += 1 在多线程下照样不安全。
• 什么时候释放：阻塞 IO（socket 读写、time.sleep）和不少 C 扩展（如 NumPy 的部分运算）会主动释放；纯 Python 代码按切换间隔（sys.getswitchinterval()，默认 5ms）被迫让出。
• 结论：多线程能并发等 IO，但算不满多核；CPU 密集要多进程。3.13 引入实验性的 free-threaded 构建（PEP 703）；3.14 起按 PEP 779 转为官方支持，但仍是可选构建，默认发行版照样有 GIL，官方给出的单线程性能损失约 5–10%。面试里按「默认 CPython 有 GIL」答，再补一句 free-threaded 的现状。

LLM 应用一次请求 90% 以上时间在等模型 API、向量库、数据库。线程方案每个连接一个 OS 线程（MB 级栈 + 内核调度），协程方案一个线程挂几千个协程，每个只是堆上一个状态对象——这就是 FastAPI、各类 Agent 框架默认 async 的原因。协程通用原理（有栈 / 无栈、切换成本）见 协程原理，这里只讲 asyncio 的具体机制。

2. 三个对象：coroutine / Task / Future#

async def fetch(i):
    await asyncio.sleep(1)
    return i

async def main():
    # 串行：共 3 秒。await 一个协程 = 直接进去跑，不产生并发
    for i in range(3):
        await fetch(i)

    # 并发：共 1 秒。create_task 把三个协程都交给事件循环
    tasks = [asyncio.create_task(fetch(i)) for i in range(3)]
    results = [await t for t in tasks]

易踩：create_task 返回的 Task 要自己留引用。事件循环对 Task 只持弱引用，不存起来的「后台任务」可能中途被 GC 回收（官方文档明确提醒）。

3. await 的挂起与恢复：事件循环到底在干嘛#

事件循环本质是一个 while True，每轮做三件事：

1. 用 selector（Linux 上是 epoll，macOS 上是 kqueue）等 IO 就绪，超时时间取「最近一个定时器还有多久到期」——机制见 IO多路复用；
2. 把到期的定时器（call_later，小顶堆）和就绪 IO 对应的回调放进 ready 队列；
3. 依次执行 ready 队列里的回调。

await 的挂起链路：

sequenceDiagram
    participant L as 事件循环
    participant T as Task
    participant C as 协程(状态机)
    participant F as Future(socket读)
    L->>T: 执行 Task.__step
    T->>C: coro.send(None)
    C->>F: await 一个未完成的 Future
    F-->>T: 把 Future 自己 yield 上来
    T->>F: add_done_callback(Task.__wakeup)
    T-->>L: 本轮返回，循环去跑别的 Task
    Note over L: epoll 报告 socket 可读
    L->>F: set_result(data)
    F->>L: call_soon(Task.__wakeup)
    L->>T: 下一轮执行 __step
    T->>C: send(data)，从 await 处继续

两个推论：

• 只有 await 点才会切换。两个 await 之间的代码对其他协程是原子的，这是单线程协程不太需要锁的原因——但跨 await 的「先读后写」照样有竞态，需要 asyncio.Lock。
• 协程没办法被抢占。一个协程不 await 就一直霸占线程，其他几千个连接全部卡住。

4. 并发编排：gather / TaskGroup / 超时#

async def multi_search(query: str):
    sem = asyncio.Semaphore(5)                 # 限制同时打出去的请求数，防止打爆下游/触发限流

    async def one(src):
        async with sem:
            return await search(src, query)

    try:
        async with asyncio.timeout(10):        # 3.11+；老版本用 asyncio.wait_for
            async with asyncio.TaskGroup() as tg:
                tasks = [tg.create_task(one(s)) for s in SOURCES]
    except* SearchError as eg:                 # ExceptionGroup 用 except* 拆
        log.warning("检索失败，已取消其余请求: %s", eg.exceptions)  # TaskGroup 一个失败就全取消，拿不到部分结果
        raise
    return [t.result() for t in tasks]

近几个版本的变化（以官方 What’s New 为准）：

• 3.11：TaskGroup、asyncio.timeout()、except*；asyncio.wait() 不再接受裸协程。
• 3.12：wait_for 改为基于 asyncio.timeout() 实现；新增 eager_task_factory，协程在创建 Task 时先同步执行，不阻塞就直接完成、不进事件循环。
• 3.13：as_completed 支持异步迭代。
• 3.14：create_task 可以透传任意关键字参数给 Task 构造器；新增 python -m asyncio ps <PID> / pstree <PID> 查看运行中进程的任务和 await 调用链，以及 asyncio.capture_call_graph() / print_call_graph()；事件循环 policy 体系被弃用。

5. 取消与 CancelledError 的传播#

• task.cancel() 不是立刻杀死，而是在该任务下一次 await 处抛出 CancelledError。超时（asyncio.timeout / wait_for）内部就是 cancel。
• 3.8 起 CancelledError 继承 BaseException，except Exception 抓不到它——这是故意的，防止业务代码误吞。
• 正确写法：清理放 finally；非要 except CancelledError 做收尾，做完必须 raise。吞掉取消会让外层超时「失效」：外层以为已经取消，任务却还在跑。
• asyncio.shield(aw) 保护内层不被外层取消（比如「扣费结算」这一步不允许半途被掐），但外层的 await 依然会收到 CancelledError。
• 取消向下传播：取消一个 Task，它正在 await 的子 Task / gather 也会被取消；反过来子任务被取消不会取消父任务，TaskGroup 也一样；TaskGroup 只在子任务抛出普通异常时取消兄弟任务和父任务。

async def call_llm_with_cleanup():
    stream = await client.open_stream()
    try:
        async for chunk in stream:
            yield chunk
    except asyncio.CancelledError:
        metrics.inc("llm_cancelled")
        raise                                  # 一定要重新抛出
    finally:
        await stream.aclose()                  # 连接必须还回去

6. 最常见的坑：阻塞调用卡死事件循环#

loop = asyncio.get_running_loop()
# IO 型同步函数：丢到默认线程池（默认 max_workers = min(32, CPU数+4)）
data = await asyncio.to_thread(legacy_sdk.query, q)
# CPU 型：线程池没用（有 GIL），要进程池
chunks = await loop.run_in_executor(proc_pool, parse_pdf, path)

排查手段：开 debug 模式（PYTHONASYNCIODEBUG=1 或 asyncio.run(main(), debug=True)），单个回调执行超过 loop.slow_callback_duration（默认 0.1 秒）会打 warning，能直接定位是哪段代码没 await。3.14 起还可以对运行中的进程执行 python -m asyncio pstree <PID>，看哪些任务卡在哪个 await 上。

7. asyncio vs 多线程 vs 多进程#

生产上常见组合：uvicorn 多 worker 进程（吃多核）× 每进程一个事件循环（吃 IO 并发）× 少量线程池（兼容同步库）。

8. 和 Java 的对照#

asyncio 是单线程加事件循环的并发模型。先说为什么用它：CPython 有 GIL，同一时刻只有一个线程执行字节码，多线程跑不满多核，但 LLM 应用大部分时间在等模型 API 和数据库，属于 IO 密集，一个线程挂几千个协程比开几千个线程省得多。机制上分三个对象：调用 async 函数得到 coroutine，本身不执行；Task 负责驱动协程；Future 是结果占位。协程 await 一个没完成的 Future 时，这个 Future 被一路 yield 到 Task，Task 给它挂一个唤醒回调后返回，事件循环就去跑别的任务；epoll 报告 IO 就绪后 Future 被 set_result，回调把 Task 重新排进 ready 队列，协程从 await 处继续。并发编排上，gather 一个失败其余不取消，3.11 的 TaskGroup 会取消兄弟任务并抛 ExceptionGroup，超时用 asyncio.timeout。取消是在下一个 await 点抛 CancelledError，它是 BaseException，清理放 finally，捕获了必须重新抛，否则外层超时会失效。最大的坑是阻塞调用：requests、同步数据库驱动、time.sleep 放进 async 函数会卡死整个循环，要换异步库，或者用 to_thread 丢线程池、CPU 活丢进程池。还有一个隐蔽的坑：第三方库如果吞了 CancelledError，wait_for 这类基于取消的超时会静默失效，这时可以用 asyncio.wait 自己判时间、处理 pending。结合项目时可以讲：整轮任务的超时加在哪一层、哪些工具可以并发、遇到过的取消或超时失效问题是怎么定位的。

追问方向：

• “有 GIL 为什么还需要 asyncio.Lock？” → GIL 只保护解释器内部状态；asyncio 在两个 await 之间是原子的，但「读余额 → await 查库 → 写余额」跨了 await，别的协程能插进来，所以跨 await 的临界区要 asyncio.Lock。
• “create_task 和直接 await 协程有什么区别？” → 直接 await 是进入协程同步地跑到底，没有并发；create_task 把协程注册进事件循环，当前协程继续往下走，下一个 await 点后它就开始跑了。
• “gather 里一个任务抛异常，其他任务怎么样？” → 默认异常立刻抛给调用方，其他任务不取消、继续在后台跑，容易留下孤儿任务；要么 return_exceptions=True 全收，要么改用 TaskGroup 自动取消兄弟任务。
• “为什么 wait_for 超时了任务还在跑？” → 超时靠 cancel 实现，被取消的协程如果 except CancelledError 后没重新抛（或第三方库吞了），取消就失效了；排查看有没有吞 BaseException 的 except，必要时用 asyncio.wait(timeout=) 自己判时间并处理 pending。
• “async def 里调用了一个同步 SDK，怎么发现和修？” → debug 模式下慢回调超过 0.1 秒会打 warning；修法是 await asyncio.to_thread(sdk.call, ...)，CPU 型改 run_in_executor(ProcessPoolExecutor)。
• “to_thread 能解决 CPU 密集吗？” → 不能，线程池里的纯 Python 计算照样受 GIL 限制，只是不卡事件循环而已，总吞吐不涨；CPU 密集要进程池或者交给释放 GIL 的 C 扩展。
• “ContextVar 在子任务里 set 了，父任务看得到吗？” → 看不到。Task 创建时拷贝当前 context，子任务的 set 只改自己那份；要跨任务汇总，就传一个可变对象（如 dict）进去，或者在父任务里收集返回值。
• “asyncio 和 Java 虚拟线程最大的区别？” → 虚拟线程是有栈的，写同步代码、JVM 在阻塞 IO 时自动卸载；asyncio 是无栈的，只能在显式 await 处挂起，同步阻塞库不会自动让出，所以有函数染色问题。
• “怎么限制同时打给 LLM 的并发数？” → asyncio.Semaphore(n) 包住调用；多进程/多副本部署时每进程各一个信号量，全局限流要放 Redis（令牌桶）。

易错：

• ❌ “用了 async 就并发了” → 顺序 await 三个协程还是串行，必须 create_task / gather / TaskGroup 才并发。
• ❌ “except Exception 能兜住所有异常” → CancelledError 是 BaseException，这恰好是对的；反过来写 except BaseException 或裸 except: 会吞掉取消。
• ❌ “多线程在 Python 里没用” → 包装同步 IO 库时线程依然有效，因为阻塞 IO 会释放 GIL。