Python在同步方法中執(zhí)行協(xié)程方法的實現(xiàn)
更新時間：2025年09月12日 08:50:44 作者：無風聽海
Python同步方法無法直接執(zhí)行協(xié)程,因其返回協(xié)程對象需事件循環(huán)調(diào)度,五種解決方案涵蓋不同場景：asyncio.run()適用于腳本,loop.run()需顯式控制,nest_asyncio解決嵌套問題但不推薦生產(chǎn),線程安全方法用于框架內(nèi)任務(wù),線程池運行協(xié)程為通用推薦方案
一、問題描述：同步方法無法直接執(zhí)行協(xié)程
在 Python 中，協(xié)程（async def 函數(shù)）返回的是一個 協(xié)程對象，它不會自動執(zhí)行，必須被事件循環(huán)調(diào)度執(zhí)行。
而 同步方法中既不能使用 await，也沒有運行事件循環(huán)，因此直接調(diào)用協(xié)程會導致如下問題：
async def my_coroutine():
return 42
def sync_func():
result = my_coroutine() # ? 這里只是得到 coroutine 對象
print(result)
# 輸出類似：<coroutine object my_coroutine at 0x...>
更嚴重的是：
如果你嘗試手動運行協(xié)程，但事件循環(huán)已經(jīng)在運行（如 Jupyter Notebook、某些異步框架如 FastAPI），會拋出異常：
RuntimeError: This event loop is already running
二、解決方案總覽
| 方案編號 | 方法 | 適用環(huán)境 | 是否阻塞主線程 | 可嵌套性 |
|---|---|---|---|---|
| 1 | asyncio.run() | 腳本 / CLI 程序主函數(shù) | ? 阻塞 | ? 不能嵌套 |
| 2 | loop.run_until_complete() | 事件循環(huán)未運行的場景 | ? 阻塞 | ? 不能嵌套 |
| 3 | nest_asyncio | Jupyter / 教學用途 | ? 阻塞 | ? 可以嵌套 |
| 4 | asyncio.run_coroutine_threadsafe() | 異步框架 / 多線程后臺任務(wù) | ? 阻塞 | ? 安全 |
| 5 | 線程池中運行事件循環(huán) | 通用同步環(huán)境 | ? 阻塞 | ? 安全 |
三、詳細方案分析
方案一：使用asyncio.run()
示例
import asyncio
async def async_func():
await asyncio.sleep(1)
return "result"
def sync_func():
result = asyncio.run(async_func())
print(result)
場景適用
- 腳本、命令行程序的主入口函數(shù)
- 保證事件循環(huán)尚未在運行
注意
- 不能嵌套使用，否則會拋出錯誤（如在 Jupyter 中使用）
RuntimeError: asyncio.run() cannot be called from a running event loop
方案二：使用loop.run_until_complete()
示例
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)
場景適用
- 桌面程序、腳本中希望顯式控制事件循環(huán)
- 環(huán)境中事件循環(huán)尚未在運行
注意
- 如果事件循環(huán)已在運行，會報錯：
RuntimeError: This event loop is already running
方案三：使用nest_asyncio允許事件循環(huán)嵌套
示例
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)
場景適用
- Jupyter Notebook
- 異步框架中臨時執(zhí)行協(xié)程（教學、調(diào)試）
注意
- 通過 monkey-patching 修改了事件循環(huán)行為，不推薦在生產(chǎn)環(huán)境使用
方案四：使用asyncio.run_coroutine_threadsafe()（線程安全）
示例
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)
場景適用
- 后臺線程管理事件循環(huán)
- 異步任務(wù)調(diào)度框架中（如 aiohttp、FastAPI）提交任務(wù)到獨立的 loop
- 多線程環(huán)境下與 asyncio 集成
注意
- 需要手動管理線程和事件循環(huán)生命周期
方案五：在線程池中運行事件循環(huán)（推薦通用方法）
示例
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()
場景適用
- 在任何同步環(huán)境中執(zhí)行協(xié)程（即使已有事件循環(huán)在運行）
- GUI 程序、Web 框架、Jupyter 等
優(yōu)點
- 安全、無嵌套沖突
- 與主線程隔離
- 線程池易于集成到現(xiàn)有項目
四、總結(jié)對比表（按推薦優(yōu)先級）
| 方法 | 是否可嵌套 | 是否阻塞 | 推薦使用場景 |
|---|---|---|---|
| asyncio.run() | ? | ? 是 | 普通腳本、CLI 程序 |
| loop.run_until_complete() | ? | ? 是 | 明確控制事件循環(huán)、服務(wù)初始化 |
| nest_asyncio | ? | ? 是 | Jupyter、教學（?? 不用于生產(chǎn)） |
| run_coroutine_threadsafe() | ? | ? 是 | 后臺線程任務(wù)調(diào)度，框架內(nèi)部 |
| 線程池運行協(xié)程 | ? | ? 是 | ? 通用方法，兼容所有同步環(huán)境 |
最佳實踐建議
- 推薦通用封裝：
def run_async(coro):
try:
loop = asyncio.get_running_loop()
except RuntimeError:
return asyncio.run(coro)
else:
# 如果事件循環(huán)正在運行，使用線程池運行
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor() as pool:
return pool.submit(lambda: asyncio.run(coro)).result()
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
今天小編就為大家分享一篇對Django的restful用法詳解(自帶的增刪改查)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-08-08
Python跨文件調(diào)用函數(shù)的五種實用方法
在開發(fā)Python項目時,90%的開發(fā)者都會遇到這樣的困境：代碼越寫越長,功能越來越亂,最后變成難以維護的"意大利面條式代碼",本文將手把手教你通過模塊化編程,讓代碼結(jié)構(gòu)清晰、可維護性強,并深入解析5種跨文件調(diào)用函數(shù)的實用方法,需要的朋友可以參考下2025-05-05
win7下 python3.6 安裝opencv 和 opencv-contrib-python解決 cv2.xfeat
這篇文章主要介紹了win7下 python3.6 安裝opencv 和 opencv-contrib-python解決 cv2.xfeatures2d.SIFT_create() 的問題,需要的朋友可以參考下2019-10-10
這篇文章主要介紹了Flask URL傳參與視圖映射的實現(xiàn)方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習吧2023-03-03
Python實現(xiàn)將Excel轉(zhuǎn)換為高保真圖片
在數(shù)據(jù)處理與自動化的辦公場景中,將 Excel 表格轉(zhuǎn)換為高保真的圖片是一項常見且極具挑戰(zhàn)性的需求,本文將深入探討如何利用 Spire.XLS for Python 將 Excel 工作表轉(zhuǎn)換為圖片格式,感興趣的小伙伴可以了解下2026-02-02
這篇文章主要介紹了Python類定義和類繼承詳解,本文講解了類的私有屬性、類的方法、私有的類方法、類的專有方法、類的定義、類的單繼承、類的多繼承等內(nèi)容,需要的朋友可以參考下2015-05-05
Pytorch中的Tensorboard與Transforms搭配使用
這篇文章主要介紹了Pytorch中的Tensorboard與Transforms搭配使用，主要是結(jié)合了前兩篇文章的的一個小練習，感興趣的小伙伴可以來練習一下，希望對你的學習有所幫助2021-12-12
今天小編就為大家分享一篇解決Pycharm 導入其他文件夾源碼的2種方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-02-02
PyCharm添加Anaconda中的虛擬環(huán)境Python解釋器出現(xiàn)Conda?executable?is?not
這篇文章主要給大家介紹了關(guān)于PyCharm添加Anaconda中的虛擬環(huán)境Python解釋器出現(xiàn)Conda?executable?is?not?found錯誤的解決辦法,文中通過圖文介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2023-02-02
詞頻統(tǒng)計是指在文本中計算每個詞出現(xiàn)的次數(shù)。這篇文章主要帶大家了解一下Python實現(xiàn)中文詞頻統(tǒng)計的方法，感興趣的小伙伴可以了解一下2023-02-02