async def my_coroutine():
    return 42

def sync_func():
    result = my_coroutine()  # ? 這里只是得到 coroutine 對(duì)象
    print(result)

# 輸出類似：<coroutine object my_coroutine at 0x...>

import asyncio

async def async_func():
    await asyncio.sleep(1)
    return "result"

def sync_func():
    result = asyncio.run(async_func())
    print(result)

RuntimeError: asyncio.run() cannot be called from a running event loop

import asyncio

async def async_func():
    await asyncio.sleep(1)
    return "result"

def sync_func():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(async_func())
    print(result)

RuntimeError: This event loop is already running

pip install nest_asyncio

import asyncio
import nest_asyncio

nest_asyncio.apply()

async def async_func():
    await asyncio.sleep(1)
    return "nested result"

def sync_func():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(async_func())
    print(result)

import asyncio
import threading

async def async_func():
    await asyncio.sleep(1)
    return "from thread-safe"

def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

def sync_func():
    loop = asyncio.new_event_loop()
    threading.Thread(target=start_loop, args=(loop,), daemon=True).start()

    future = asyncio.run_coroutine_threadsafe(async_func(), loop)
    result = future.result()  # 阻塞直到完成
    print(result)

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def async_func():
    await asyncio.sleep(1)
    return "from thread pool"

def run_coroutine_in_thread(coro):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    return loop.run_until_complete(coro)

def sync_func():
    with ThreadPoolExecutor() as executor:
        future = executor.submit(run_coroutine_in_thread, async_func())
        result = future.result()
        print(result)

sync_func()

def run_async(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    else:
        # 如果事件循環(huán)正在運(yùn)行，使用線程池運(yùn)行
        from concurrent.futures import ThreadPoolExecutor
        with ThreadPoolExecutor() as pool:
            return pool.submit(lambda: asyncio.run(coro)).result()