如何在同步代码中获取asyncio.Task.result且避免阻塞事件循环？

• 你在运行事件循环的同线程内调用loop.create_task(second_async()).result()时，刚创建的任务还在事件循环的调度队列中未执行，此时同步调用.result()会直接阻塞当前线程（也就是事件循环所在线程），事件循环无法继续调度任务执行，second_async永远不会运行，任务始终处于Pending状态，调用.result()自然会抛出InvalidStateError。
• 原生asyncio不允许在已运行的事件循环内嵌套调用run_until_complete，所以你不能直接用该方法执行协程拿结果。

使用第三方库nest_asyncio给事件循环打补丁，支持嵌套运行协程，完美兼容你的两种调用场景：跨线程调用sync、事件循环线程内的async代码调用sync。

实现步骤 #

1. 安装依赖：
   pip install nest-asyncio
2. 修改后的完整代码：

import asyncio
import nest_asyncio

event_loop = asyncio.new_event_loop()
# 给指定事件循环打补丁，支持嵌套运行
nest_asyncio.apply(event_loop)
asyncio.set_event_loop(event_loop)

async def second_async():
    # 你的异步逻辑
    print("I'm here")
    return 123

def sync():
    try:
        # 判断当前线程是否存在正在运行的事件循环
        current_loop = asyncio.get_running_loop()
    except RuntimeError:
        # 无运行中循环，走跨线程调用逻辑
        return asyncio.run_coroutine_threadsafe(second_async(), loop=event_loop).result()
    # 当前运行的就是目标循环，直接嵌套执行协程拿结果
    if current_loop is event_loop:
        return current_loop.run_until_complete(second_async())
    # 当前运行的是其他事件循环，仍走跨线程调用
    return asyncio.run_coroutine_threadsafe(second_async(), loop=event_loop).result()

async def first_async():
    res = sync()
    print(f"获取到结果: {res}")

event_loop.run_until_complete(first_async())

方案说明 #

• nest_asyncio修改了原生asyncio事件循环的实现逻辑，允许在已运行的事件循环内调用run_until_complete，此时会临时调度执行传入的协程直到完成，再回到之前的执行上下文，不会中断原有事件循环的运行。
• 如果你无法引入第三方依赖，只能将sync函数改为异步接口，或把后续对结果的处理逻辑改为回调形式，但不符合你要求的同步接口集成需求，因此该方案是当前场景下的最优解。

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