如何在批量异步任务完成后后台执行异步任务且不阻塞循环？

核心思路是用asyncio.create_task()将单独的异步任务提交到事件循环后台执行，无需等待它完成即可进入下一轮循环。同时必须处理任务的异常，避免未捕获异常引发的警告或程序问题。

具体实现步骤 #

• 每轮先通过asyncio.gather()执行X个任务并等待全部完成，确保当前轮次的核心任务都处理完毕。
• 调用asyncio.create_task()启动单独的异步任务：这个方法会把任务注册到当前事件循环，只要事件循环处于运行状态，任务就会在后台自动调度执行，不需要当前协程等待它结束。
• 给单独任务套一层异常捕获包装：如果任务可能抛出异常，未处理的话会在任务被回收时触发asyncio.exceptions.TaskDestroyed警告，因此必须在任务内部或包装器中捕获并处理异常。

代码示例 #

import asyncio

async def x_task(task_id):
    # 模拟每轮的X个任务逻辑
    await asyncio.sleep(0.2)
    print(f"X任务 {task_id} 执行完成")

async def standalone_task():
    # 模拟需要后台执行的单独任务逻辑
    await asyncio.sleep(0.6)
    print("后台单独任务执行完成")

async def safe_standalone_task():
    # 异常捕获包装器，避免未处理异常触发警告
    try:
        await standalone_task()
    except Exception as e:
        print(f"后台任务异常: {str(e)}")

async def main():
    total_rounds = 3  # 循环总轮数
    x_per_round = 2   # 每轮的X个任务数量

    for round_idx in range(total_rounds):
        print(f"\n=== 第 {round_idx+1} 轮开始 ===")
        # 创建并等待当前轮的X个任务完成
        current_x_tasks = [x_task(i) for i in range(x_per_round)]
        await asyncio.gather(*current_x_tasks)
        
        # 启动后台任务，无需等待即可进入下一轮
        asyncio.create_task(safe_standalone_task())
        
        print(f"=== 第 {round_idx+1} 轮结束，进入下一轮 ===")

if __name__ == "__main__":
    asyncio.run(main())

关键细节说明 #

• asyncio.create_task()是核心：它会将任务加入事件循环的任务队列，事件循环会在处理完当前协程的挂起点后，自动调度执行后台任务，完全不阻塞主循环的推进。
• 异常处理不可省略：即使你认为任务不会出错，也要加上异常捕获——未处理的异常会导致程序运行时出现警告，甚至在极端情况下影响事件循环的稳定性。
• 任务生命周期：只要事件循环处于运行状态，后台任务就会执行到完成；如果主程序退出（比如asyncio.run(main())执行完毕），未完成的后台任务会被直接终止。

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