如何在Jupyter Notebook中用Asyncio实现嵌套异步调用？

你当前的代码已经实现了单个示例内RF与BM模型的并行运行，但无法让多个示例的处理流程异步并行。核心需求是：

• 每个示例的处理流程（并行跑RF+BM → 对比结果）独立异步执行
• 单个示例内必须等RF和BM的任务全部完成后，才能执行对比逻辑

先修正原代码的语法问题 #

• 函数定义def compare(RF, BM)和for循环for (i,n) in enumerate(Ex_cases)末尾缺少冒号:
• 创建BM任务时错误传入了RF模型，应该是Run_sim(BM, Ex_cases[i])

修改后的完整代码 #

import asyncio
from concurrent.futures import ThreadPoolExecutor  # 处理同步阻塞的模型运行逻辑

# 初始化线程池，根据CPU核心数调整max_workers
executor = ThreadPoolExecutor(max_workers=4)

def compare(rf_res, bm_res):
    # 实现你的结果对比逻辑，示例为检查结果是否一致
    is_consistent = rf_res == bm_res
    return f"示例对比结果: {'一致' if is_consistent else '不一致'}"

def run_model(model, case):
    # 这里是你的模型同步运行逻辑，比如模拟耗时操作
    import time
    time.sleep(1)
    return f"{model}_output_{case}"

async def run_sim(model, case):
    # 用线程池包装同步模型函数，避免阻塞asyncio事件循环
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, run_model, model, case)
    return result

async def process_single_case(rf_model, bm_model, case):
    # 单个示例的完整异步处理流程
    # 1. 并行启动当前示例的两个模型任务
    rf_task = asyncio.create_task(run_sim(rf_model, case))
    bm_task = asyncio.create_task(run_sim(bm_model, case))
    
    # 2. 等待当前示例的两个模型任务全部完成
    rf_res, bm_res = await asyncio.gather(rf_task, bm_task)
    
    # 3. 执行结果对比并输出
    comparison_result = compare(rf_res, bm_res)
    print(comparison_result)
    return comparison_result

async def main():
    # 示例集、参考模型、基准模型（假设已定义）
    Ex_cases = ["Example1", "Example2", "Example3", "Example4", "Example5"]
    RF = "Reference_Model"
    BM = "Benchmark_Model"
    
    # 为所有示例创建异步处理任务
    case_tasks = [
        asyncio.create_task(process_single_case(RF, BM, case))
        for case in Ex_cases
    ]
    
    # 等待所有示例的处理任务完成
    await asyncio.gather(*case_tasks)

# 在Jupyter Notebook中直接运行主函数
await main()

关键改动说明 #

• 封装单示例逻辑：把每个示例的"并行跑模型→对比结果"流程封装成process_single_case异步函数，确保单个示例内的依赖逻辑（必须等两个模型跑完再对比）被严格执行。
• 批量创建示例任务：在main函数中一次性为所有示例创建处理任务，通过asyncio.gather实现多示例的异步并行处理，大幅提升整体运行效率。
• 处理同步阻塞函数：如果你的run_model是同步阻塞类型（比如CPU密集计算或同步IO操作），必须用线程池/进程池包装，否则会阻塞asyncio的事件循环，无法实现真正的并行。

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