李宏彦讲Python异步：3个API变更避坑指南

版本升级后 API 全变了，代码直接报错？这是很多开发者在重构老项目时的噩梦。李宏彦在深入剖析 Python 异步编程演进时，特别强调了一个核心观点：不要盲目追逐新特性，而要理解底层调度逻辑的变迁。这篇避坑指南，就是为你梳理从 Python 3.4 到 3.12 之间，asyncio 模块那些“悄悄”改变的关键点，帮你把那些因为版本差异导致的“灵异现象”一次性解决。

入口定位：为什么你的 async 代码突然卡死了？

很多学员问我：“老师，我代码明明加了 async/await，为什么跑起来比同步还慢？”或者“为什么 await 一个函数有时候返回协程对象，有时候直接返回值？”

这通常不是代码写错了，而是你运行的 Python 版本和调用的 API 语义发生了变化。以 asyncio.run() 为例，在 Python 3.7 之前，我们习惯用 loop.run_until_complete()。但 3.7 引入了 asyncio.run() 作为官方推荐入口。更隐蔽的坑在于 事件循环的生命周期管理。

在 Python 3.8 之前，如果你在一个已经运行的事件循环中再次尝试启动新的循环，或者在子线程中错误地复用主线程的循环，很容易出现 RuntimeError: This event loop is already running。而在新版本中，asyncio 对“当前线程是否有活动循环”的检查更加严格。

这里有个真实的案例：某培训机构学员在 Django 项目中集成异步任务，使用了 asyncio.run() 在视图函数中执行。在 Python 3.7 测试环境正常，升级到 3.10 后，一旦并发请求增多，就频繁出现 RuntimeError: asyncio.run() cannot be called from a running event loop。根本原因是 Django 3.2+ 引入了异步视图支持，底层已经启动了事件循环，而学员的代码又在同一个上下文中强制启动了另一个循环。

避坑要点： 在 Web 框架中使用异步，务必确认框架是否已经管理了事件循环。如果框架已启动循环，你只能 await 协程，绝不能再次调用 asyncio.run()。

核心片段：剖析 `asyncio.run()` 的底层实现

为了搞清楚版本差异，我们直接看 Python 3.11 源码中 asyncio/runners.py 的核心逻辑。这段代码决定了 asyncio.run() 如何接管主线程的控制权。

# 源码片段：Python 3.11 asyncio/runners.py
# 注意：这是简化版，仅展示核心调度逻辑class Runner:def __init__(self, debug=None):self._state = RunnerState.IDLEself._loop = Noneself._main_task = Noneself._context = Noneself._set_event_loop = Truedef run(self, coro, *, context=None):# 1. 状态检查：防止重入if self._state != RunnerState.IDLE:raise RuntimeError("Runner is already running")# 2. 初始化事件循环：这是版本差异的关键点# 在 3.8+ 中，run() 会创建一个新的 IsolatedLoop# 而在旧版本中，可能直接复用 get_event_loop()loop = events.new_event_loop()self._loop = loopself._set_event_loop = Trueevents.set_event_loop(loop)try:# 3. 创建主任务self._main_task = loop.create_task(coro)# 4. 运行直到主任务完成# 这里使用了 run_forever 的变体逻辑loop.run_until_complete(self._main_task)# 5. 收集所有待处理任务（关键避坑点）# 3.8+ 版本会强制检查是否有未完成的 Taskall_tasks = tasks.all_tasks(loop)pending = [t for t in all_tasks if not t.done()]if pending:# 抛出异常，防止资源泄露raise RuntimeError(f"Unfinished tasks: {pending}")finally:# 6. 清理循环self._cleanup()return self._main_task.result()

逐行解读：

状态锁机制：if self._state != RunnerState.IDLE 是防止嵌套调用的第一道防线。在 Python 3.8 之前，这种检查分散在 run_until_complete 中，容易绕过。现在集中管理，更安全。
events.new_event_loop()：这是最大的变化。旧代码常用 loop = asyncio.get_event_loop()。如果当前线程没有循环，它会创建一个；如果有，就返回现有的。这导致在多线程或 Web 框架中，get_event_loop() 可能返回一个已关闭或不属于当前线程的循环。而 asyncio.run() 强制创建新循环，隔离性更好。
tasks.all_tasks(loop)：这是 3.7 引入的 API。它返回当前循环中所有未完成的 Task。很多新手忘记 await 某些后台任务，导致程序退出时这些任务被静默取消，数据不一致。新版本通过 raise RuntimeError 强制暴露这个问题，虽然让开发期报错变多，但避免了生产环境的数据静默丢失。
self._cleanup()：确保循环关闭、上下文清理。旧版本中，如果异常中断，循环可能处于“半开”状态，导致后续 asyncio.get_event_loop() 返回一个坏掉的循环。

关键洞察： asyncio.run() 的设计哲学是“一次性、隔离、强制清理”。它不适合长生命周期的应用（如服务器），只适合脚本、测试或一次性任务。如果你的应用需要长期运行事件循环，请手动管理 loop 的生命周期。

设计思想：从“全局单例”到“显式依赖”

理解源码后，我们需要看透设计思想的转变。早期 asyncio 依赖全局变量 event_loop，这是一种隐式依赖。这种设计在单线程、单循环场景下没问题，但在多线程、多循环（如 Jupyter Notebook、Web 框架）场景下，灾难频发。

Python 3.10 及以后的官方文档明确建议：避免使用 asyncio.get_event_loop()，因为它在行为上具有歧义。

对比表格：API 演进与行为差异

API	Python 3.7 及以前	Python 3.8 - 3.10	Python 3.11+
`get_event_loop()`	返回当前线程循环，若无则创建	返回当前线程循环，若无则发出 DeprecationWarning 并创建	若当前线程无循环，抛出 DeprecationWarning，建议用 `new_event_loop`
`run_until_complete()`	直接运行协程	需要传入 loop 参数	同左，但更强调显式传递
`asyncio.run()`	不存在	引入，创建新循环，强制清理	稳定，推荐用于顶层入口
`Task.cancel()`	仅设置标志位	设置标志位，需 `await` 才能生效	优化了取消传播，确保 `CancelledError` 正确抛出

核心设计思想变化：

显式优于隐式：强制开发者明确指定在哪个循环中运行任务，避免跨线程/跨上下文混淆。
失败快速（Fail Fast）：未完成的 Task 不再静默忽略，而是抛出异常。这符合“显式错误优于隐式错误”的原则。
上下文隔离：每个 asyncio.run() 调用拥有独立的事件循环和上下文，避免状态污染。

对于培训机构学员，理解这一点至关重要：不要迷信“自动”功能，要理解底层资源的生命周期。在面试中，能讲清楚 get_event_loop 为什么被弃用，以及 asyncio.run 如何管理循环，是高级 Python 开发者的基本素养。

手写简化版：构建一个安全的异步执行器

为了彻底掌握这些概念，我们手写一个简化版的 safe_async_run，模拟 asyncio.run() 的核心行为，并加入额外的错误处理。

import asyncio
import threading
import tracebackdef safe_async_run(coro, *, debug=False, timeout=None):"""简化版的异步执行器，模拟 asyncio.run 的核心逻辑适用于教学演示，不建议在生产环境直接替换 asyncio.run"""# 1. 检查是否已在事件循环中try:current_loop = asyncio.get_running_loop()raise RuntimeError("Cannot call safe_async_run() from a running event loop. ""Use await instead.")except RuntimeError:# 没有运行中的循环，继续pass# 2. 创建新的事件循环loop = asyncio.new_event_loop()asyncio.set_event_loop(loop)result = Nonetry:# 3. 创建任务并附加异常处理task = loop.create_task(coro)# 4. 设置超时（可选）if timeout:task = asyncio.wait_for(task, timeout=timeout)# 5. 运行循环loop.run_until_complete(task)result = task.result()# 6. 检查是否有残留任务remaining = asyncio.all_tasks(loop)if remaining:for t in remaining:t.cancel()loop.run_until_complete(asyncio.gather(*remaining, return_exceptions=True))raise RuntimeError(f"Leftover tasks found: {remaining}")except Exception as e:# 7. 记录详细错误信息error_msg = f"Async task failed: {e}\n{traceback.format_exc()}"print(error_msg)raisefinally:# 8. 清理循环try:# 关闭所有未关闭的资源loop.close()except Exception as e:print(f"Error closing loop: {e}")finally:asyncio.set_event_loop(None)  # 清除当前线程的循环引用return result# 测试用例
async def sample_task():await asyncio.sleep(1)return "Hello, Async!"if __name__ == "__main__":# 正常执行result = safe_async_run(sample_task())print(f"Result: {result}")# 测试异常async def failing_task():await asyncio.sleep(1)raise ValueError("Something went wrong")try:safe_async_run(failing_task())except ValueError as e:print(f"Caught expected error: {e}")

代码解析：

asyncio.get_running_loop()：这是 3.7+ 的 API，用于检查当前线程是否已有运行中的循环。比 get_event_loop() 更精确，因为它只返回正在运行的循环，而不关心是否存在但未运行的循环。
asyncio.set_event_loop(None)：在清理阶段，将当前线程的循环引用设为 None。这防止后续代码意外获取到一个已关闭的循环。这是很多新手忽略的细节，导致调试时出现“幽灵错误”。
asyncio.wait_for：用于实现超时控制。在生产环境中，任何异步操作都应有超时限制，防止无限期挂起。
残留任务处理：通过 asyncio.all_tasks 检查是否有未完成的 Task。如果有，强制取消并等待其结束。这确保了资源被正确释放。

教学建议： 让学员在 Jupyter Notebook 和 Django 项目中分别运行这段代码，观察 get_running_loop() 的行为差异。在 Jupyter 中，由于 IPython 已启动事件循环，get_running_loop() 会返回一个循环，因此 safe_async_run 会抛出异常，提示使用 await。这正好演示了为什么在交互式环境中不能直接使用 asyncio.run()。

应用场景：在 Web 框架中正确集成异步

理论讲完，落地才是关键。以下是在 Flask 和 FastAPI 中集成异步任务的最佳实践。

场景 1：Flask 中调用异步函数

Flask 本质上是同步框架，但它支持在请求处理中调用异步函数。错误做法是直接在视图函数中调用 asyncio.run()，因为 Flask 可能在多线程环境下运行，导致循环冲突。

正确做法： 使用线程池将异步任务隔离到独立线程中执行。

from flask import Flask
import asyncio
import concurrent.futuresapp = Flask(__name__)# 创建全局线程池
executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)def run_async_in_thread(coro):"""在独立线程中运行异步协程"""def _run():# 在新线程中，没有运行中的循环，可以安全创建loop = asyncio.new_event_loop()asyncio.set_event_loop(loop)try:return loop.run_until_complete(coro)finally:loop.close()asyncio.set_event_loop(None)return _run@app.route('/async-endpoint')
def async_endpoint():# 提交异步任务到线程池future = executor.submit(run_async_in_thread(asyncio.sleep(2) or print("Done")))result = future.result(timeout=5)  # 同步等待结果，带超时return {"message": "Async task completed"}

关键点：

线程隔离：每个异步任务运行在独立线程中，拥有独立的事件循环，避免与 Flask 的主线程冲突。
超时控制：future.result(timeout=5) 确保请求不会无限期挂起。
循环清理：loop.close() 和 set_event_loop(None) 确保资源释放。

场景 2：FastAPI 中定义异步端点

FastAPI 原生支持异步，这是其核心优势。

from fastapi import FastAPI
import httpxapp = FastAPI()@app.get("/fetch-data")
async def fetch_data():# 直接 await 异步函数，无需手动管理循环async with httpx.AsyncClient() as client:response = await client.get("https://httpbin.org/get")return response.json()

关键点：

自动管理：FastAPI 框架负责创建和管理事件循环，开发者只需 await。
非阻塞 I/O：httpx.AsyncClient 是非阻塞的，相比同步的 requests，能显著提高并发性能。
避免阻塞调用：在异步端点中，绝不能调用同步阻塞函数（如 time.sleep、requests.get），否则会阻塞整个事件循环，导致其他请求无法处理。

避坑总结：

不要混用同步和异步客户端：在异步上下文中，始终使用异步版本的库（如 httpx 而非 requests，aiomysql 而非 pymysql）。
不要手动创建循环：在框架管理的上下文中，信任框架的循环管理，不要自己 new_event_loop()。
始终设置超时：任何网络请求、数据库操作都应有超时限制。