asyncio.run()后get_event_loop()报错及同步上下文调用异步函数方案咨询

asyncio.run 的设计定位是单线程异步程序的顶层入口方法，它的内部执行逻辑固定为：创建全新的事件循环、将其设置为当前线程的绑定循环、运行传入的协程任务、待协程执行结束后主动关闭该循环，同时清理当前线程绑定的事件循环实例。
因此调用完 asyncio.run 后，主线程已经没有关联的活跃事件循环，此时调用 asyncio.get_event_loop() 自然会抛出 RuntimeError: There is no current event loop in thread 'MainThread' 报错。
每次调用 asyncio.run 后手动设置新事件循环的用法属于临时hack，不符合asyncio的设计预期，容易出现资源泄漏、跨循环任务调度异常等问题。

根据不同的使用场景，优先选择以下符合设计规范的方案：

场景1：全程序仅需一次异步逻辑调度 #

如果所有异步任务可以统一收拢调度，不需要中途穿插同步逻辑零散提交任务，直接将所有异步逻辑封装到同一个入口协程中，全程仅调用一次 asyncio.run 即可，无需手动管理事件循环：

import asyncio

async def asyncfunction(n):
    print(f'before sleep in asyncfunction({ n })')
    await asyncio.sleep(1)
    print(f'after sleep in asyncfunction({ n })')
    return f'result of asyncfunction({ n })'

def callback(r):
    print(f'inside callback, got: {r}')

async def main():
    # 所有异步逻辑统一在此调度
    r0 = await asyncfunction(0)
    callback(r0)
    print('sync code following asyncfunction(0)')
    
    r1 = await asyncfunction(1)
    callback(r1)
    print('sync code following asyncfunction(1)')
    
    # 后台任务也在此提交
    task2 = asyncio.create_task(wrapper(2))
    print('sync code following create_task(2)')
    
    task3 = asyncio.create_task(wrapper(3))
    print('sync code following create_task(3)')
    
    # 等待所有后台任务完成
    await asyncio.gather(task2, task3)

async def wrapper(n):
    r = await asyncfunction(n)
    callback(r)

if __name__ == "__main__":
    asyncio.run(main())

该方案是asyncio官方推荐的标准用法，适配绝大多数普通异步脚本场景。

场景2：必须在同步逻辑中零散调用异步函数 #

如果业务逻辑无法重构为统一的异步入口，需要在同步代码执行过程中多次调用异步函数、提交后台任务，可手动维护全局唯一的事件循环实例，全程复用该循环完成所有异步调度：

import asyncio

# 全局初始化一次事件循环，全程复用
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

# 封装同步调用异步函数的通用方法
def run_async(coro):
    return loop.run_until_complete(coro)

# 业务逻辑调用
r0 = run_async(asyncfunction(0))
callback(r0)
print('sync code following asyncfunction(0)')

r1 = run_async(asyncfunction(1))
callback(r1)
print('sync code following asyncfunction(1)')

# 提交后台任务直接复用全局循环
loop.create_task(wrapper(2))
print('sync code following create_task(2)')

loop.create_task(wrapper(3))
print('sync code following create_task(3)')

# 所有逻辑执行完成后，等待剩余任务跑完再关闭循环
loop.run_until_complete(asyncio.gather(*asyncio.all_tasks(loop)))
loop.close()

该方案避免了反复创建销毁事件循环的开销，也不会出现循环状态异常的问题。

特殊场景：已有活跃事件循环的环境 #

如果代码运行在自带活跃事件循环的环境（如Jupyter Notebook、异步Web服务）中，无法调用asyncio.run也无法手动创建新循环，可使用nest_asyncio库对现有循环打补丁，支持嵌套运行异步任务，该方案属于兼容性处理，普通脚本场景不推荐使用。

内容的提问来源于stack exchange，提问作者luochen1990