Python在同步方法中執(zhí)行協(xié)程方法的實現(xiàn)
更新時間：2025年09月12日 08:50:44 作者：無風聽海
Python同步方法無法直接執(zhí)行協(xié)程,因其返回協(xié)程對象需事件循環(huán)調度,五種解決方案涵蓋不同場景：asyncio.run()適用于腳本,loop.run()需顯式控制,nest_asyncio解決嵌套問題但不推薦生產,線程安全方法用于框架內任務,線程池運行協(xié)程為通用推薦方案
一、問題描述：同步方法無法直接執(zhí)行協(xié)程
在 Python 中，協(xié)程（async def 函數(shù)）返回的是一個 協(xié)程對象，它不會自動執(zhí)行，必須被事件循環(huán)調度執(zhí)行。
而 同步方法中既不能使用 await，也沒有運行事件循環(huán)，因此直接調用協(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() | 異步框架 / 多線程后臺任務 | ? 阻塞 | ? 安全 |
| 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é)程（教學、調試）
注意
- 通過 monkey-patching 修改了事件循環(huá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)
- 異步任務調度框架中（如 aiohttp、FastAPI）提交任務到獨立的 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)有項目
四、總結對比表（按推薦優(yōu)先級）
| 方法 | 是否可嵌套 | 是否阻塞 | 推薦使用場景 |
|---|---|---|---|
| asyncio.run() | ? | ? 是 | 普通腳本、CLI 程序 |
| loop.run_until_complete() | ? | ? 是 | 明確控制事件循環(huán)、服務初始化 |
| nest_asyncio | ? | ? 是 | Jupyter、教學（?? 不用于生產） |
| run_coroutine_threadsafe() | ? | ? 是 | 后臺線程任務調度，框架內部 |
| 線程池運行協(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)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
使用Python實現(xiàn)企業(yè)微信的自動打卡功能
下面就通過Python程序來實現(xiàn)自動打卡，原理很簡單，用Python設置定時任務，然后通過adb操作手機，完成打卡功能，感興趣的朋友跟隨小編一起看看吧2019-04-04
在VSCode中配置Python開發(fā)環(huán)境的詳細教程
Visual Studio Code（簡稱VSCode）以其強大的功能和靈活的擴展性,成為了許多開發(fā)者的首選,本文將詳細介紹如何在VSCode中配置Python開發(fā)環(huán)境,需要的朋友可以參考下2025-04-04
這篇文章主要介紹了python中plt.cm.Paired的用法說明，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
這篇文章主要介紹了Pytorch 實現(xiàn)常用正則化的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
Python的Flask框架中的Jinja2模板引擎學習教程
這篇文章主要介紹了Python的Flask框架中的Jinja2模板引擎學習教程,Jinja2模板引擎的用法也是Flask的Web開發(fā)中的重要知識,需要的朋友可以參考下2016-06-06
對于Python程序員來說,處理彈出窗口似乎并不是一個常見的任務,這篇文章主要給大家介紹了關于Python圖形化界面基礎篇之如何使用彈出窗口和對話框的相關資料,需要的朋友可以參考下2024-03-03
Codeium插件可以自動建議、解釋代碼、生成注釋和根據(jù)注釋生成代碼等,本文主要介紹了PyCharm無法登陸Codeium的解決方法,感興趣的可以了解一下2023-11-11
Python Django模板之模板過濾器與自定義模板過濾器示例
這篇文章主要介紹了Python Django模板之模板過濾器與自定義模板過濾器,結合實例形式分析了Django框架模板過濾器與自定義模板過濾器相關功能、原理、使用方法及相關操作注意事項,需要的朋友可以參考下2019-10-10
這篇文章主要介紹了如何將一個CSV格式的文件分割成兩個CSV文件，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-07-07
Python實現(xiàn)個人微信號自動監(jiān)控告警的示例
今天小編就為大家分享一篇Python實現(xiàn)個人微信號自動監(jiān)控告警的示例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-07-07