本文清晰梳理 asyncio 中事件循环（Event Loop）的创建方式（asyncio.run() 与 asyncio.new_event_loop() 的区别）和启动机制（run_until_complete() 与 run_forever()），明确各 API 的职责层级、调用时机及典型使用场景，并辅以可运行示例。
本文清晰梳理 asyncio 中事件循环（event loop）的创建方式（`asyncio.run()` 与 `asyncio.new_event_loop()` 的区别）和启动机制（`run_until_complete()` 与 `run_forever()`），明确各 api 的职责层级、调用时机及典型使用场景，并辅以可运行示例。
在 Python 的 asyncio 框架中，事件循环是异步 I/O 的核心调度器，所有协程、任务（Task）、Future 和回调都依赖它执行。但初学者常混淆“创建”与“获取”、“启动”与“运行”的概念。关键在于理解 asyncio 提供了高阶（high-level）与低阶（low-level）两套 API：前者面向应用开发，封装细节；后者面向框架或高级控制需求，提供精细操作能力。
✅ 事件循环的创建方式
-
asyncio.run(coro)（推荐，高阶 API）
这是最安全、最常用的入口。它内部自动完成三件事：- 调用 asyncio.new_event_loop() 创建一个全新事件循环（若传入 loop_factory，则使用该工厂）；
- 通过 asyncio.set_event_loop() 将其设为当前线程的默认循环；
- 调用 loop.run_until_complete(coro) 启动并等待协程完成。
✅ 优点：自动管理生命周期，避免循环复用/泄漏；❌ 缺点：每次调用都会新建循环，不可重复使用。
asyncio.new_event_loop()（低阶 API）
纯创建函数，返回一个未启动、未绑定的 AbstractEventLoop 实例（如 SelectorEventLoop）。
⚠️ 注意：它不设置为当前线程默认循环，也不启动——需手动调用 asyncio.set_event_loop(loop) 和 loop.run_until_complete(...)。asyncio.get_event_loop()（⚠️ 已不推荐用于新代码）
它不创建新循环，而是返回当前线程的“已设置”事件循环（即 get_event_loop_policy().get_event_loop()）。
❗ 在 asyncio.run() 之外首次调用时，可能返回一个未运行、未配置的默认循环（行为因 Python 版本而异），易引发 RuntimeError（如 “no running event loop”）。自 Python 3.10+ 起，官方明确建议优先使用 asyncio.run()。
▶️ 事件循环的启动方式
一旦拥有事件循环对象（loop），启动方式取决于执行目标：
| 方法 | 行为 | 典型用途 |
|---|---|---|
| loop.run_until_complete(coro) | 运行循环直至 coro 返回或抛出异常，然后关闭循环 | 执行单个主协程（如 main()） |
| loop.run_forever() | 持续运行循环，直到显式调用 loop.stop() | 构建长期服务（如 WebSocket 服务器、自定义调度器） |
? 提示：asyncio.run() 底层正是调用 loop.run_until_complete()，因此它天然适合一次性脚本或 CLI 工具；而 run_forever() 更适合嵌入式或框架级应用。
? 完整对比示例
以下两个等效实现均获取模拟远程数据，突出高阶与低阶 API 的差异：
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
立即学习“Python免费学习笔记（深入）”；
✅ 高阶写法（推荐）
import asyncio
async def get_data():
data = [{"name": "Jane", "age": 19}, {"name": "John", "age": 24}]
await asyncio.sleep(1) # 模拟 I/O 延迟
return data
if __name__ == "__main__":
result = asyncio.run(get_data()) # 自动创建 + 启动 + 清理
print(result)⚙️ 低阶写法（需手动管理）
import asyncio
async def get_data():
data = [{"name": "Jane", "age": 19}, {"name": "John", "age": 24}]
await asyncio.sleep(1)
return data
def main():
# 手动创建并设置循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
# 创建任务并运行
task = loop.create_task(get_data())
loop.run_until_complete(task)
print(task.result())
finally:
loop.close() # 显式清理资源（重要！）
if __name__ == "__main__":
main()⚠️ 关键注意事项
- 永远不要在 asyncio.run() 内部再调用 asyncio.run() —— 会引发 RuntimeError: asyncio.run() cannot be called from a running event loop。
- 使用低阶 API 时，务必在 loop.close() 前确保循环已停止（loop.is_closed() 为 True），否则可能引发资源泄漏。
- 在多线程环境中，每个线程应独立创建并管理自己的事件循环（asyncio.new_event_loop() + set_event_loop()），主线程默认循环不跨线程共享。
- Python 3.12+ 引入 asyncio.Runner 类，进一步封装低阶流程，适用于需多次运行不同协程的场景（如测试框架），可作为 asyncio.run() 的增强替代。
掌握事件循环的创建与启动逻辑，是写出健壮、可维护异步 Python 代码的基础。对绝大多数应用，请坚持使用 asyncio.run()；仅当需要深度定制（如嵌入 GUI 主循环、混合同步/异步逻辑）时，才谨慎选用低阶 API。
