Quart框架中如何让create_task创建的后台任务完成后再退出?
解决方案
你的代码存在几个关键问题,导致无法正常等待后台任务完成后再退出:
_is_initialised初始为False,run()里的循环根本不会执行stop()方法过早打印"Shutdown has completed",但此时后台任务还没真正结束- 没有保存后台任务的引用,无法在 shutdown 阶段等待它完成
以下是修正后的完整实现:
import asyncio from quart import Quart class Application: def __init__(self): self._is_initialised: bool = False self._shutdown_completed: bool = False self._shutdown_requested: bool = False async def run(self) -> None: # 等待初始化完成后再进入业务循环 while not self._is_initialised: await asyncio.sleep(0.1) while not self._shutdown_requested: try: # 这里替换成你的实际数据分析、报告生成逻辑 await self._main_loop() await asyncio.sleep(0.1) except Exception as e: # 捕获任务异常,避免单个错误导致后台任务崩溃 print(f"后台任务异常: {str(e)}") print("后台任务开始执行退出前清理...") # 可在此添加清理逻辑:关闭IO连接、保存临时数据等 # await self._cleanup() self._shutdown_completed = True print("后台任务已完成退出") def stop(self) -> None: print("Shutting down...") self._shutdown_requested = True async def _main_loop(self): # 模拟IO密集型业务任务 print("正在执行后台数据分析任务...") await asyncio.sleep(2) # 替换为实际耗时操作 app = Quart(__name__) SERVICE_APP = Application() background_task = None # 保存后台任务引用,用于后续等待 @app.before_serving async def startup() -> None: global background_task SERVICE_APP._is_initialised = True # 创建后台任务并保存引用 background_task = asyncio.create_task(SERVICE_APP.run()) @app.after_serving async def shutdown() -> None: SERVICE_APP.stop() # 等待后台任务完全执行完毕 if background_task is not None: await background_task if __name__ == "__main__": app.run()
关键修改说明:
- 保存任务引用:在
startup中保留create_task返回的任务对象,这样在shutdown时可以通过await等待任务结束 - 修复初始化逻辑:在
startup中设置_is_initialised = True,确保后台任务的业务循环能正常启动 - 调整关闭流程:
stop()仅设置关闭标志,后台任务会自行完成剩余操作并打印退出信息 - 异常防护:在任务循环中添加异常捕获,避免单个业务错误导致后台任务直接崩溃
- 强制等待退出:在
after_serving钩子中,调用stop()后等待后台任务完成所有逻辑,再结束应用
现在按下Ctrl+C时,Quart会先触发关闭流程,设置后台任务的终止标志,等待任务完成剩余操作(包括清理)后,才会真正退出应用。
内容的提问来源于stack exchange,提问作者Swatcat
相关产品推荐
相关产品推荐

