Python如何从同步方法执行asyncio方法并阻塞完成？避免async/await扩散

这个需求完全可行，核心是利用Python asyncio库的同步入口函数，在同步代码中桥接异步任务，无需修改下游业务逻辑的同步代码。

报错原因分析 #

你遇到的"An asyncio.Future, a coroutine or an awaitable is required"错误，通常是因为传递给异步等待方法（如asyncio.gather/asyncio.wait）的不是可等待对象（协程、Future或Task），比如直接传入了函数对象而非调用函数生成的协程实例。

具体实现代码 #

1. 定义异步函数 #

import asyncio

async def async_task1():
    await asyncio.sleep(2)
    print("异步任务1执行完成")
    return "任务1结果"

async def async_task2():
    await asyncio.sleep(1)
    print("异步任务2执行完成")
    return "任务2结果"

2. 同步代码中执行并等待异步任务 #

通过asyncio.run()（Python 3.7+支持）作为同步与异步的桥接点，它会自动创建、运行并销毁事件循环，直到传入的协程执行完毕才返回，自然阻塞后续同步代码：

def execute_async_tasks():
    # 生成协程对象列表（必须调用异步函数，不能直接传函数本身）
    coroutines = [async_task1(), async_task2()]
    # 收集所有任务并等待完成，返回结果列表
    return asyncio.run(asyncio.gather(*coroutines))

# 同步执行异步任务，直到所有任务完成才继续
task_results = execute_async_tasks()
print(f"所有异步任务完成，结果: {task_results}")

3. 下游同步业务逻辑 #

直接编写普通同步函数即可，无需任何async/await关键字：

def downstream_sync_logic(results):
    print("开始执行下游同步业务逻辑...")
    for idx, res in enumerate(results):
        print(f"处理第{idx+1}个任务结果: {res}")

# 调用下游逻辑，直接使用异步任务的结果
downstream_sync_logic(task_results)

特殊场景处理 #

如果你的代码运行在已有事件循环的异步环境（如FastAPI、aiohttp的请求上下文），不能使用asyncio.run()，可改用当前事件循环的run_until_complete方法：

def execute_async_tasks_in_running_loop():
    loop = asyncio.get_running_loop()
    coroutines = [async_task1(), async_task2()]
    return loop.run_until_complete(asyncio.gather(*coroutines))

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