在运行中的事件循环内从同步函数调用异步函数的方案

同步函数boo()会在运行中的事件循环内被调用，需从异步函数foo(arg1, arg2)获取数据，但boo()无法改为异步函数，必须实现从boo()内调用异步函数并等待其完成后继续执行。

最小复现示例 #

以下是最贴近实际场景的代码：

import time
import asyncio

async def work_for_data():
    time.sleep(3)
    return 42

# 同步函数，需要调用异步函数
def get_number():
    return asyncio.get_event_loop().run_until_complete(work_for_data())

async def get_data():
    return get_number()

async def run():
    loop = asyncio.get_event_loop()
    task = asyncio.create_task(get_data())
    loop.run_until_complete(task)

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

执行后报错：

File "./minimal_example.py", line 9, in get_number
    return asyncio.get_event_loop().run_until_complete(work_for_data())
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/Cellar/python@3.11/3.11.3/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 629, in run_until_complete
    self._check_running()
  File "/usr/local/Cellar/python@3.11/3.11.3/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 588, in _check_running
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running

已尝试的无效解决方案 #

尝试1 #

data = asyncio.run(foo(arg1, arg2))

报错：

data = asyncio.run(foo(arg1, arg2))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/root/.pycharm_helpers/pydevd_asyncio/pydevd_nest_asyncio.py", line 143, in run
    loop.run_until_complete(task)
  File "uvloop/loop.pyx", line 1511, in uvloop.loop.Loop.run_until_complete
  File "uvloop/loop.pyx", line 1504, in uvloop.loop.Loop.run_until_complete
  File "uvloop/loop.pyx", line 1377, in uvloop.loop.Loop.run_forever
  File "uvloop/loop.pyx", line 518, in uvloop.loop.Loop._run
RuntimeError: this event loop is already running.

尝试2 #

loop = asyncio.get_event_loop()
data = loop.run_until_complete(foo(arg1, arg2))

报错：

data = loop.run_until_complete(foo(arg1, arg2))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "uvloop/loop.pyx", line 1511, in uvloop.loop.Loop.run_until_complete
  File "uvloop/loop.pyx", line 1504, in uvloop.loop.Loop.run_until_complete
  File "uvloop/loop.pyx", line 1377, in uvloop.loop.Loop.run_forever
  File "uvloop/loop.pyx", line 518, in uvloop.loop.Loop._run
RuntimeError: this event loop is already running.

尝试3 #

loop = asyncio.get_running_loop()
with ThreadPoolExecutor() as executor:
    future = executor.submit(lambda: asyncio.run_coroutine_threadsafe(foo(arg1, arg2), loop).result())
    data = future.result()

执行到future.result()时解释器卡死。

尝试4 #

loop = asyncio.get_event_loop()
future = asyncio.Future()

def callback(task):
    if task.exception():
        future.set_exception(task.exception())
    else:
        future.set_result(task.result())

task = asyncio.run_coroutine_threadsafe(foo(arg1, arg2), loop)
task.add_done_callback(callback)

result = task.result()  ## 此处卡死
return result

执行到task.result()时解释器卡死。

可行解决方案 #

方案1：使用nest_asyncio允许事件循环嵌套 #

nest_asyncio可以修改asyncio的默认行为，允许在已运行的事件循环中再次调用run_until_complete或asyncio.run。

1. 安装依赖库：

pip install nest_asyncio

2. 修改后的完整代码：

import time
import asyncio
import nest_asyncio

# 应用补丁，开启嵌套事件循环支持
nest_asyncio.apply()

async def work_for_data():
    time.sleep(3)
    return 42

def get_number():
    return asyncio.get_event_loop().run_until_complete(work_for_data())

async def get_data():
    return get_number()

async def run():
    task = asyncio.create_task(get_data())
    await task

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

此方案无需大幅修改原有代码，仅需添加补丁即可解决冲突。

方案2：在新线程中运行异步函数 #

若不想引入第三方库，可通过创建新线程，在新线程中启动独立事件循环执行异步函数，同步函数等待线程完成后获取结果。

修改后的完整代码：

import time
import asyncio
from threading import Thread

async def work_for_data():
    time.sleep(3)
    return 42

def get_number():
    result = None
    error = None

    def run_async_task():
        nonlocal result, error
        try:
            # 在新线程中创建独立事件循环
            loop = asyncio.new_event_loop()
            result = loop.run_until_complete(work_for_data())
        except Exception as e:
            error = e
        finally:
            loop.close()

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

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

async def get_data():
    return get_number()

async def run():
    task = asyncio.create_task(get_data())
    await task

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

此方案通过线程隔离事件循环，避免了当前线程事件循环已运行的冲突，完全基于Python标准库实现。

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