李宏彦讲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.IDLE
self._loop = None
self._main_task = None
self._context = None
self._set_event_loop = True
def 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 = loop
self._set_event_loop = True
events.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+ 版本会强制检查是否有未完成的 Task
all_tasks = tasks.all_tasks(loop)
pending = [t for t in all_tasks if not t.done()]
if pending:
# 抛出异常，防止资源泄露
raise RuntimeError(fUnfinished 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 traceback
def 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 = None
try:
# 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(fLeftover tasks found: {remaining})
except Exception as e:
# 7. 记录详细错误信息
error_msg = fAsync task failed: {e}\n{traceback.format_exc()}
print(error_msg)
raise
finally:
# 8. 清理循环
try:
# 关闭所有未关闭的资源
loop.close()
except Exception as e:
print(fError 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(fResult: {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(fCaught 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.futures
app = 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 httpx
app = 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()。
始终设置超时：任何网络请求、数据库操作都应有超时限制。
结尾互动
从 asyncio.run() 的源码剖析，到 Web 框架中的实际应用，我们看到了 Python 异步编程从“隐式魔法”到“显式控制”的演进。李宏彦强调，理解这些底层机制，比记住多少 API 更重要。版本升级带来的 API 变化，本质上是设计思想的迭代，目的是让代码更健壮、更可预测。
现在，轮到你思考一下：在你的项目中，你更常用 asyncio.run() 还是手动管理事件循环？遇到过哪些因为版本升级导致的“灵异”问题？评论区交流，我们一起避坑。 
长河君 · 2026/9/22 17:01:55 · 阅读 1.2k 李宏彦讲Python异步：3个API变更避坑指南
李宏彦讲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.IDLE
self._loop = None
self._main_task = None
self._context = None
self._set_event_loop = True
def 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 = loop
self._set_event_loop = True
events.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+ 版本会强制检查是否有未完成的 Task
all_tasks = tasks.all_tasks(loop)
pending = [t for t in all_tasks if not t.done()]
if pending:
# 抛出异常，防止资源泄露
raise RuntimeError(fUnfinished 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 traceback
def 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 = None
try:
# 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(fLeftover tasks found: {remaining})
except Exception as e:
# 7. 记录详细错误信息
error_msg = fAsync task failed: {e}\n{traceback.format_exc()}
print(error_msg)
raise
finally:
# 8. 清理循环
try:
# 关闭所有未关闭的资源
loop.close()
except Exception as e:
print(fError 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(fResult: {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(fCaught 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.futures
app = 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 httpx
app = 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()。
始终设置超时：任何网络请求、数据库操作都应有超时限制。
结尾互动
从 asyncio.run() 的源码剖析，到 Web 框架中的实际应用，我们看到了 Python 异步编程从“隐式魔法”到“显式控制”的演进。李宏彦强调，理解这些底层机制，比记住多少 API 更重要。版本升级带来的 API 变化，本质上是设计思想的迭代，目的是让代码更健壮、更可预测。
现在，轮到你思考一下：在你的项目中，你更常用 asyncio.run() 还是手动管理事件循环？遇到过哪些因为版本升级导致的“灵异”问题？评论区交流，我们一起避坑。