快捷導(dǎo)航

在Python同步方法中調(diào)用異步方法不阻塞主流程的幾種方案

更新時(shí)間：2025年03月14日 08:21:47 作者：oliver.chau

這篇文章主要介紹了在Python同步方法中調(diào)用異步方法不阻塞主流程的幾種方案,包括使用asyncio.create_task()、threading和concurrent.futures,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下

方案 1：使用 asyncio.create_task()（推薦）

如果你的代碼 運(yùn)行在事件循環(huán)中（如 Flask + Quart 或 FastAPI），可以使用：

import asyncio

async def save_category_icon(category_id=0, category_name=None):
    await asyncio.sleep(2)  # 模擬異步任務(wù)
    print(f"Icon saved for category {category_id} - {category_name}")

def edit_category():
    loop = asyncio.get_event_loop()
    loop.create_task(save_category_icon(category_id=1, category_name="Test"))
    print("Main process continues...")  # 不會(huì)等待異步任務(wù)完成

edit_category()

? 優(yōu)點(diǎn)：

不會(huì)阻塞主線程
任務(wù)會(huì)在后臺(tái)執(zhí)行
適用于已運(yùn)行的 asyncio 事件循環(huán)（FastAPI、Quart 等）

方案 2：使用 threading（適用于 Flask 同步環(huán)境）

如果你的代碼在 Flask（同步框架）中運(yùn)行，推薦用 threading 運(yùn)行異步方法：

import threading
import asyncio

def run_async_task():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(save_category_icon(category_id=1, category_name="Test"))

def edit_category():
    threading.Thread(target=run_async_task, daemon=True).start()
    print("Main process continues...")  # 不會(huì)等待異步任務(wù)完成

edit_category()

? 優(yōu)點(diǎn)：

不會(huì)阻塞 Flask 主進(jìn)程
在 Flask 等同步環(huán)境下仍可運(yùn)行異步任務(wù)
daemon=True 確保進(jìn)程退出時(shí)線程自動(dòng)關(guān)閉

方案 3：使用 concurrent.futures（適用于高并發(fā)情況）

如果你的異步任務(wù)是 CPU 密集型 或者需要更好的線程池管理，可以使用 concurrent.futures：

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

def run_async_task():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(save_category_icon(category_id=1, category_name="Test"))

def edit_category():
    executor.submit(run_async_task)  # 提交任務(wù)到線程池
    print("Main process continues...")

edit_category()

? 優(yōu)點(diǎn)：

適用于高并發(fā)任務(wù)
可以復(fù)用 ThreadPoolExecutor，提高性能
不阻塞主進(jìn)程

選哪個(gè)方案？

方案	適用場(chǎng)景	優(yōu)點(diǎn)
`asyncio.create_task()`	代碼已經(jīng)運(yùn)行在事件循環(huán)（FastAPI、Quart）	輕量級(jí)，非阻塞
`threading.Thread()`	Flask 或 Django（同步框架）	適用于 Flask，后臺(tái)執(zhí)行任務(wù)
`concurrent.futures.ThreadPoolExecutor`	高并發(fā)或多個(gè)異步任務(wù)	適用于 CPU 密集型任務(wù)

總結(jié)

到此這篇關(guān)于在Python同步方法中調(diào)用異步方法不阻塞主流程的幾種方案的文章就介紹到這了,更多相關(guān)Python同步方法調(diào)用異步方法內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: