如何在未知是否已有运行中事件循环时从同步函数调用异步函数？

针对你提出的「在同步函数中调用异步函数，兼容有无运行中事件循环」的需求，这里给出一个可行的better_asyncio_run实现：

import asyncio
from threading import Thread

def better_asyncio_run(coro):
    try:
        # 检查当前线程是否有活跃的事件循环
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # 无活跃循环时，直接用asyncio.run启动协程
        return asyncio.run(coro)
    else:
        # 已有活跃循环时，在新线程中启动独立循环执行协程
        result = None
        def _run_coro_in_new_loop():
            nonlocal result
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            result = new_loop.run_until_complete(coro)
            new_loop.close()
        
        thread = Thread(target=_run_coro_in_new_loop)
        thread.start()
        thread.join()
        return result

核心逻辑说明 #

• 无事件循环场景：直接调用asyncio.run()，符合官方常规用法，自动创建并管理循环生命周期。
• 已有事件循环场景：由于同一线程无法同时运行两个asyncio事件循环，因此通过创建新线程，在新线程内初始化独立的事件循环来执行目标协程，等待线程结束后返回结果，实现同步阻塞直到协程完成。

验证示例 #

将你提供的测试代码替换为上述函数后，可正常运行：

async def coro1():
    print("coro1 执行完成")

def function1():
    better_asyncio_run(coro1())

def function2():
    function1()
    better_asyncio_run(coro1())

async def coro2():
    function2()

if __name__ == '__main__':
    better_asyncio_run(coro2())

运行后会输出两次coro1 执行完成，不会触发「事件循环已运行」的错误。

关于方案的补充说明 #

为什么不能在同一线程内启动第二个事件循环？因为asyncio的事件循环与线程强绑定，同一线程同一时间只能有一个循环处于运行状态，强行调用会抛出RuntimeError。线程隔离是当前唯一可行的兼容方案。

至于「把所有代码改成异步」的建议，确实不符合快速原型开发的高效需求，这个方案可以在不重构原有同步代码结构的前提下，实现异步协程的调用。

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