应先检测事件循环是否运行，若已运行则降级为同步执行，否则调用asyncio.run()；需统一超时、异常和日志，避免连接池差异导致行为不一致。
asyncio.run() 里抛出 RuntimeError 怎么办
直接在非异步上下文（比如普通函数、同步脚本入口）里调用 asyncio.run() 是安全的，但如果你已经身处一个运行中的事件循环里（比如 Flask/Gunicorn 环境、Jupyter、或被其他库启动过 loop），再调用它就会报 RuntimeError: asyncio.run() cannot be called from a running event loop。这不是代码写错了，是环境冲突。
这时候不能硬扛，得主动识别并降级：
- 先用
asyncio.get_event_loop()尝试获取当前 loop - 如果
is_running()返回True，说明不能新建 loop，必须走同步路径 - 别依赖
asyncio._get_running_loop()—— 它在 Python 3.7+ 才有，且下划线开头是私有 API，容易在 patch 版本里静默失效
如何统一接口：一个函数，自动选 async 或 sync 执行
核心不是“封装 try/except”，而是提前判断执行环境。把异步逻辑和同步逻辑拆成两个独立函数，再用一个调度器路由：
def fetch_data(url):
try:
loop = asyncio.get_event_loop()
if loop.is_running():
raise RuntimeError("loop already running")
return asyncio.run(_fetch_async(url))
except RuntimeError:
return _fetch_sync(url)
async def _fetch_async(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.text()
def _fetch_sync(url):
import requests
return requests.get(url).text
注意：aiohttp 和 requests 的返回值结构不一致（比如 headers 处理、编码推断），降级后要对齐字段，否则上层调用会崩。
asyncio.create_task() 在 sync 环境里直接报错
create_task() 必须在运行中的 event loop 里调用，否则会立刻抛 RuntimeError: no running event loop。它不像 asyncio.run() 那样自带兜底逻辑。
如果你的代码里混用了 task 创建和条件降级，最容易漏掉的是「task 对象本身没 await，但创建就失败」：
- 不要在函数开头就无条件
asyncio.create_task(...) - 改用
asyncio.ensure_future()？不行，它同样要求 loop 存在 - 真正安全的做法：只在确认
loop.is_running() == True时才建 task；否则直接调同步版函数 - 别试图用
asyncio.new_event_loop()+set_event_loop()手动补环 —— 多线程下极易和框架（如 FastAPI）的 loop 管理冲突，GIL 不救你
降级后日志和超时行为不一致怎么办
异步 HTTP 客户端（aiohttp）默认没有全局超时，而 requests 默认永不超时（实际受系统 socket 超时限制）。降级后表面能跑，但线上可能卡死半小时才报错。
必须显式对齐关键行为：
-
aiohttp.ClientTimeout(total=30)和requests.get(url, timeout=30)要数值一致 - 异常类型不同：
aiohttp.ClientErrorvsrequests.RequestException，统一包装成自定义异常（比如NetworkError）再往上抛 - 日志里必须打标：
"fallback to sync mode due to running loop"，否则排查时根本分不清哪次请求走的是哪条路
最麻烦的其实是连接池：aiohttp 的 TCPConnector 和 requests 的 urllib3.PoolManager 完全不共享状态，降级后并发数、复用率、DNS 缓存全都不一样。别指望“看起来结果一样”就认为行为等价。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
