如何在Jupyter同步环境中正确调度asyncio代码并获取结果

问题原因 #

在Jupyter/IPython环境中，事件循环始终在主线程运行。你当前的代码在检测到循环运行时，直接在主线程调用future.result()阻塞等待结果，但主线程被阻塞后，事件循环无法处理你提交的call_many_async协程任务，导致死锁——协程永远无法完成，主线程永远等不到结果，最终程序挂起。

修复后的代码 #

我们可以将等待future结果的逻辑放到另一个独立线程中执行，避免阻塞主线程的事件循环：

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def call_many_async(fun, many_kwargs):
    return await asyncio.gather(*[asyncio.to_thread(fun, **kwargs) for kwargs in many_kwargs])
    
def call_many(fun, many_kwargs):
    loop = asyncio.get_event_loop()
    if loop.is_running():
        # 将等待逻辑放到独立线程，避免阻塞主线程事件循环
        def wait_for_result_in_thread():
            future = asyncio.run_coroutine_threadsafe(call_many_async(fun, many_kwargs), loop)
            return future.result()
        
        # 用线程池执行等待逻辑
        with ThreadPoolExecutor(max_workers=1) as executor:
            return executor.submit(wait_for_result_in_thread).result()
    else:
        return loop.run_until_complete(call_many_async(fun, many_kwargs))

工作原理 #

1. 当事件循环已运行时，创建子线程，在子线程中调用future.result()阻塞等待结果
2. 主线程继续运行事件循环，处理call_many_async中的协程任务
3. 协程任务完成后，子线程的future.result()拿到结果并返回，主线程通过executor.submit(...).result()获取最终结果

测试验证 #

用你提供的测试代码在Jupyter中运行：

import time
def something_complex(param) -> int:
    print(f"call started with {param=}")
    time.sleep(0.1) # 模拟耗时API调用
    print("call ended")
    return 3 # 返回结果

results = call_many(something_complex, ({"param": i} for i in range(1, 5)))
print(results) # 输出: [3, 3, 3, 3]

此时程序会正常输出任务开始/结束信息，并返回结果，不会出现挂起。

内容的提问来源于stack exchange，提问作者Matěj Račinský