如何正确运行含嵌套异步任务的同步代码的Python异步函数？

我想要运行一个包含同步代码段（该同步代码段中又嵌套异步任务）的Python异步脚本，是否可行？我的需求是执行do_async3函数内的异步任务并获取结果，执行一些同步任务后，将其包装在另一个与之前异步任务独立的异步任务中，但运行代码时出现报错。

原示例代码 #

import asyncio
import threading

async def do_async3():
    # Asynchronous task logic
    print("Async Task 3 started")
    await asyncio.sleep(2)  # Simulating asynchronous work
    print("Async Task 3 completed")

def do1():
    print("Task 1")

def do2():
    print("Task 2")

def do4():
    print("Task 4")

def do5():
    print("Task 5")

async def main():
    print("main")
    # Synchronous tasks
    do1()
    do2()

    # Asynchronous task
    await do_async3()

    # Synchronous tasks after async task
    do4()
    do5()

# Call asyncio.run(main()) from a synchronous context
def run_main_sync(loop):
    print("run_main_sync")
    asyncio.set_event_loop(loop)
    loop.run_until_complete(main())

# New asynchronous function that calls run_main_sync
async def run_main_async():
    print("start")
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    run_main_sync(loop)

# Run the asynchronous function
asyncio.run(run_main_async())

报错信息 #

(env) python proba.py
start
run_main_sync

Traceback (most recent call last):
...
RuntimeError: Cannot run the event loop while another loop is running

sys:1: RuntimeWarning: coroutine 'main' was never awaited

解决方案 #

错误原因 #

你遇到的RuntimeError是因为在一个已经运行的事件循环里，又尝试启动另一个事件循环（loop.run_until_complete(main())），这在asyncio的单线程单循环模型中是不允许的。同时你的代码逻辑绕了弯路，完全不需要手动创建新循环再同步调用。

基础实现：异步流程穿插同步代码 #

如果核心需求是在异步流程中按顺序执行同步代码、异步任务、同步代码，最简写法如下：

import asyncio

async def do_async3():
    print("Async Task 3 started")
    await asyncio.sleep(2)
    print("Async Task 3 completed")
    return "Async3 Result"  # 返回任务结果供后续使用

def do1():
    print("Task 1")

def do2():
    print("Task 2")

def do4():
    print("Task 4")

def do5():
    print("Task 5")

async def main():
    print("main")
    # 执行同步任务
    do1()
    do2()

    # 执行异步任务并获取结果
    async_result = await do_async3()
    print(f"Got async result: {async_result}")

    # 继续执行同步任务
    do4()
    do5()

# 直接启动主异步逻辑
asyncio.run(main())

进阶实现：包装为独立异步任务 #

如果确实需要把main的逻辑包装成一个独立的异步任务（比如和其他异步任务并行执行），可以用asyncio.create_task来创建独立任务：

import asyncio

async def do_async3():
    print("Async Task 3 started")
    await asyncio.sleep(2)
    print("Async Task 3 completed")
    return "Async3 Result"

def do1():
    print("Task 1")

def do2():
    print("Task 2")

def do4():
    print("Task 4")

def do5():
    print("Task 5")

async def main():
    print("main")
    do1()
    do2()
    async_result = await do_async3()
    print(f"Got async result: {async_result}")
    do4()
    do5()

async def run_main_async():
    print("start")
    # 创建独立异步任务
    task = asyncio.create_task(main())
    # 等待任务完成
    await task

asyncio.run(run_main_async())

额外提示 #

• 异步函数中可以直接调用同步函数，但同步代码会阻塞当前事件循环，直到同步代码执行完毕才会处理其他异步任务。
• 如果同步代码本身是耗时操作，不想阻塞事件循环，可以用asyncio.to_thread把同步代码放到单独线程中执行：

  # 示例：将耗时同步任务放入线程
  await asyncio.to_thread(do1)
  

内容的提问来源于stack exchange，提问作者Ádám Molnár