Python同步方法中调用异步方法实现非阻塞主流程执行策略

如果你想在 同步方法 中调用 异步方法 save_category_icon，并且不阻塞主线程，可以使用 asyncio.create_task() 或 threading/concurrent.futures 来实现。

✅ 方案 1：使用 `asyncio.create_task()`（推荐）

如果你的代码 运行在事件循环中（如 Flask + Quart 或 FastAPI），可以使用：

import asyncio

async def save_category_icon(category_id=0, category_name=None):
    await asyncio.sleep(2)  # 模拟异步任务
    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...")  # 不会等待异步任务完成

edit_category()

✅ 优点：

不会阻塞主线程

任务会在后台执行

适用于已运行的 asyncio 事件循环（FastAPI、Quart 等）

✅ 方案 2：使用 `threading`（适用于 Flask 同步环境）

如果你的代码在 Flask（同步框架）中运行，推荐用 threading 运行异步方法：

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...")  # 不会等待异步任务完成

edit_category()

✅ 优点：

不会阻塞 Flask 主进程

在 Flask 等同步环境下仍可运行异步任务

daemon=True 确保进程退出时线程自动关闭

✅ 方案 3：使用 `concurrent.futures`（适用于高并发情况）

如果你的异步任务是 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)  # 提交任务到线程池
    print("Main process continues...")

edit_category()

✅ 优点：

适用于高并发任务

可以复用 ThreadPoolExecutor，提高性能

不阻塞主进程

🎯 选哪个方案？

方案	适用场景	优点
`asyncio.create_task()`	代码已经运行在事件循环（FastAPI、Quart）	轻量级，非阻塞
`threading.Thread()`	Flask 或 Django（同步框架）	适用于 Flask，后台执行任务
`concurrent.futures.ThreadPoolExecutor`	高并发或多个异步任务	适用于 CPU 密集型任务