如何正确使用asyncio.run_coroutine_threadsafe解决future.result()阻塞问题

这类run_coroutine_threadsafe返回的Future永久卡住的问题，本质上不是API本身的缺陷，是代码逻辑触发了asyncio单线程模型的自死锁：

• 用asyncio.run(main_async())做测试时，事件循环本身就运行在主线程上
• 代码执行到async2sync分支时，检测到当前存在运行中的loop，就临时创建线程池提交任务，然后在主线程也就是loop所在的线程上，直接调用thread_future.result()做操作系统级的真阻塞等待
• 此时主线程完全被阻塞，腾不出手处理事件循环的任务调度，独立线程里通过run_coroutine_threadsafe提交给loop的协程根本得不到执行机会，独立线程也就永远等不到Future的返回值
• 两个线程互相等待对方完成，哪怕设置了3秒超时，到点也只会抛超时异常，永远拿不到真实执行结果

除此之外还有两个容易踩的坑：

• 协程对象是只能被await一次的消费型对象，跨线程透传时如果不做上下文拷贝，很容易出现变量上下文绑定错误
• 不管开多少独立线程做中转，只要在loop所在的线程上做真阻塞等待，且等待的任务依赖loop自身的调度逻辑，就一定会触发死锁

核心逻辑要做两个关键调整：

1. 先判断当前代码所在的线程，是不是事件循环本身的运行线程
2. 把所有真阻塞的等待动作，全部挪到独立于loop的线程上执行，绝对不能让loop所在的线程干阻塞的事

修正后的完整可运行代码如下：

import asyncio
from concurrent.futures import ThreadPoolExecutor
import threading

# 全局维护一个常驻等待线程池，避免每次调用都临时开线程的开销
_wait_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="async2sync-wait")

async def gather_coroutines(*coroutines):
    return await asyncio.gather(*coroutines)

def _thread_safe_wait(loop, coros):
    # 这个函数全程跑在独立的等待线程上，不占用loop的执行线程
    future = asyncio.run_coroutine_threadsafe(gather_coroutines(*coros), loop)
    return future.result(timeout=10)

def async2sync(*coroutines):
    try:
        current_loop = asyncio.get_running_loop()
    except RuntimeError:
        # 没有正在运行的事件循环，直接走asyncio.run的常规逻辑即可
        return asyncio.run(gather_coroutines(*coroutines))
    
    # 判断当前线程是不是事件循环的运行线程
    current_tid = threading.get_ident()
    if current_tid != current_loop._thread_id:
        # 当前线程和loop线程不重合：可以放心把等待逻辑扔到独立线程，等结果返回
        return _wait_pool.submit(_thread_safe_wait, current_loop, coroutines).result()
    else:
        # 当前线程就是loop线程：不能做真阻塞，不然立刻死锁
        # 直接把协程提交到当前loop，返回future对象供上层异步链路await即可
        return asyncio.ensure_future(gather_coroutines(*coroutines))

# 原有测试用例保持调整后可运行
async def some_async_task(n):
    print('Task running with n =', n)
    await asyncio.sleep(n/10)
    print('Inside coro', n)
    return list(range(n))

async def main_async():
    coro3 = some_async_task(30)
    coro1 = some_async_task(10)
    coro2 = some_async_task(20)
    results = await async2sync(coro3, coro1, coro2)
    return results

def main_sync():
    coro3 = some_async_task(30)
    coro1 = some_async_task(10)
    coro2 = some_async_task(20)
    results = async2sync(coro3, coro1, coro2)
    return results

if __name__ == '__main__':
    # 同步场景测试，正常运行
    print(main_sync())
    # 嵌套loop场景测试，不会再出现永久阻塞
    print(asyncio.run(main_async()))

如果在Jupyter Notebook场景下使用这个逻辑，因为Jupyter的事件循环本身就运行在独立的io线程上，和执行单元格代码的主线程不重合，所以async2sync的跨线程等待逻辑可以正常工作，不会触发死锁，也完全符合不新建事件循环、直接复用现有外层loop的要求。

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