Python使用asyncio实现不阻塞主流程的异步后台任务咨询

asyncio.run()是阻塞式调用，会启动全新的事件循环，直到传入的协程以及协程内创建的所有关联任务全部执行完毕才会返回，这就是代码卡在任务执行、无法提前return的根本原因。

除此之外原代码还有两个隐藏bug：

• process_background_task内循环调用asyncio.create_task后没有持有任务引用，协程执行完后这些临时创建的任务会被Python垃圾回收机制直接销毁，根本不会完整执行
• 存在多处语法错误：if(status)缺少冒号、函数传参格式错误、process_background_task定义缺少def关键字、for循环内冗余写类型标注会触发语法报错

根据你的运行场景选对应方案即可。

方案1：异步常驻服务场景（无额外依赖，性能最优） #

如果你的程序本身是基于asyncio生态的常驻服务（比如FastAPI、aiohttp这类异步Web框架），不要手动调用asyncio.run()新建事件循环，直接把后台任务提交到当前正在运行的事件循环中即可，同时做好任务引用持有避免被GC回收。

修正后代码：

async def main(args):
    transformed_data_list: List[Dict] = translate_request_to_object(args)
    status = insert_data_into_db(transformed_data)
    if status:
        # 获取当前运行中的事件循环，不新建
        loop = asyncio.get_running_loop()
        # 创建后台任务，不await就不会阻塞当前流程
        loop.create_task(process_background_task(transformed_data_list))

    # 数据库插入完成后立刻返回，不会等待后台任务
    return "data insert into db"

async def process_background_task(transformed_data_list: List[Dict]):
    task_set = set()
    for data in transformed_data_list:
        task = asyncio.create_task(heavy_computation_task(data))
        # 任务执行完成后自动从集合中移除引用
        task.add_done_callback(task_set.discard)
        task_set.add(task)
    # 让出一次事件循环调度权，保证所有子任务成功注册
    await asyncio.sleep(0)

注意：该方案仅适用于程序主事件循环常驻的场景，比如Web服务进程不会因为接口返回就退出。如果是一次性执行的脚本场景，return后进程直接终止会导致后台任务被强制中断，这类场景选方案2。

方案2：同步主流程/一次性脚本场景 #

如果main是同步函数、没有常驻的异步事件循环，单独开一个守护线程运行独立的异步事件循环即可，守护线程不会阻塞主线程返回，会随主进程正常退出。

修正后代码：

import asyncio
from threading import Thread

def _async_bg_task_runner(data_list):
    # 在线程内启动独立事件循环跑后台任务
    asyncio.run(process_background_task(data_list))

def main(args):
    transformed_data_list: List[Dict] = translate_request_to_object(args)
    status = insert_data_into_db(transformed_data)
    if status:
        # 启动守护线程跑后台任务，主线程不会等待线程执行完成
        bg_thread = Thread(
            target=_async_bg_task_runner,
            args=(transformed_data_list,),
            daemon=True
        )
        bg_thread.start()

    # 数据库插入完成后立刻返回
    return "data insert into db"

该方案下process_background_task的实现和方案1保持一致即可。

方案3：CPU密集型重计算场景优化 #

如果你的heavy_computation_task是纯CPU密集型逻辑（而非IO等待类逻辑），asyncio协程依然会占用主线程CPU资源，这种场景直接用进程池跑后台任务更合理，不会和主流程抢CPU资源：

from concurrent.futures import ProcessPoolExecutor

# 全局初始化进程池，避免每次调用重复创建进程带来的开销
process_pool = ProcessPoolExecutor(max_workers=4)

def main(args):
    transformed_data_list: List[Dict] = translate_request_to_object(args)
    status = insert_data_into_db(transformed_data)
    if status:
        for data in transformed_data_list:
            # 把CPU密集任务提交到进程池，不阻塞主流程
            process_pool.submit(heavy_computation_task_sync, data)

    return "data insert into db"

注意：该方案下重计算逻辑需要写成同步函数，不要在进程池任务内嵌套asyncio事件循环，避免出现不可预期的嵌套报错。

内容的提问来源于stack exchange，提问作者MathProblem