如何将异步函数封装为可在任意上下文正常工作的同步函数？

我完全理解你的痛点——想只实现一次异步逻辑，然后在同步和异步上下文里都能无缝调用对应的同步封装函数，不会因为事件循环的问题翻车。你遇到的核心问题本质上是：asyncio本身不支持嵌套的事件循环操作，不管是直接用asyncio.run还是尝试在运行中的循环上调用run_until_complete，都会触发RuntimeError。

先复盘下你踩过的坑 #

• 直接用asyncio.run当run函数：在纯同步上下文里没问题，但如果在异步函数（比如你的async_h）里调用封装后的同步函数（sync_g），asyncio.run会尝试创建新的事件循环，而当前线程已经有一个运行中的循环，直接报错。
• 你的run2尝试检测现有循环再用run_until_complete：但run_until_complete的设计就是要在事件循环未启动时调用，当前循环已经在运行了，自然也会报错。

给你两个可行的解决方案 #

方案一：线程隔离事件循环（无依赖、最稳妥）

思路很简单：当检测到当前有运行中的事件循环时，把协程的执行放到新线程里，在新线程中启动独立的事件循环执行协程，当前线程等待线程完成后拿到结果。这样既不会阻塞原有的事件循环，也能同步获取协程的返回值。

代码实现：

import asyncio
from threading import Thread
from typing import Coroutine, TypeVar, Any

_ReturnT = TypeVar("_ReturnT")

def run(coro: Coroutine[Any, Any, _ReturnT]) -> _ReturnT:
    try:
        # 检查当前线程是否有运行中的事件循环
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # 无运行中循环，直接用asyncio.run执行
        return asyncio.run(coro)
    else:
        # 已有运行中循环，用线程隔离执行协程
        result = None
        error = None

        def _thread_executor():
            nonlocal result, error
            try:
                # 在新线程里启动独立的事件循环
                result = asyncio.run(coro)
            except Exception as e:
                error = e

        # 启动线程并等待完成
        thread = Thread(target=_thread_executor)
        thread.start()
        thread.join()

        # 抛出执行过程中出现的异常
        if error is not None:
            raise error
        return result

用这个run函数替换你之前的实现，再跑你的测试代码：

async def async_f():
    await asyncio.sleep(0.1)
    return "async result"

def sync_f():
    return run(async_f())

def sync_g():
    print("Doing some synchronous things")
    res = sync_f()
    print("Doing some other synchronous things")
    return res

async def async_h():
    print("Doing some asynchronous things")
    res = sync_g()
    print("Doing some other asynchronous things")
    return res

# 纯同步上下文测试
print("=== 纯同步调用测试 ===")
print(sync_g())

# 异步上下文调用同步函数测试
print("\n=== 异步上下文调用测试 ===")
asyncio.run(async_h())

不管是纯同步调用sync_g，还是在异步函数async_h里调用sync_g，都能正常输出结果，不会再报错。

方案二：用nest_asyncio打补丁（简洁但有副作用）

如果你觉得线程的写法有点繁琐，也可以用第三方库nest_asyncio，它会patch asyncio的事件循环，允许嵌套执行事件循环操作。

步骤如下：

1. 先安装库：pip install nest_asyncio
2. 代码实现：

import asyncio
import nest_asyncio
from typing import Coroutine, TypeVar, Any

_ReturnT = TypeVar("_ReturnT")

# 先patch asyncio，允许嵌套事件循环
nest_asyncio.apply()

def run(coro: Coroutine[Any, Any, _ReturnT]) -> _ReturnT:
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    else:
        # 现在loop.run_until_complete可以在运行中的循环里正常执行了
        return loop.run_until_complete(coro)

⚠️ 注意：这个方案的副作用是会改变asyncio的默认行为，在涉及信号处理、子进程管理的复杂异步场景中，可能会出现不可预期的问题。如果你的代码场景比较简单，这个写法会更清爽；但如果追求稳定性，还是优先选方案一。

最后总结 #

• 无依赖、高稳定性需求：选方案一的线程隔离方式，适配所有场景。
• 代码简洁优先、场景简单：选方案二的nest_asyncio方式，代码量更少。

备注：内容来源于stack exchange，提问作者Thrastylon