FastAPI中Asyncio任务异常求助：任务阻塞与BackgroundTasks未异步

问题1：后台定时任务互相阻塞 #

核心原因

出现阻塞大概率是以下情况之一：

• 某任务包含同步阻塞代码（如time.sleep()、CPU密集计算），独占了事件循环
• 任务的死循环逻辑中没有主动让出CPU（缺少await操作）
• 优雅关闭逻辑未遵循asyncio任务的正确取消流程

解决步骤

1. 修复任务协程的非阻塞逻辑
   确保所有后台任务的核心循环中包含await操作，让事件循环有机会调度其他任务。例如修改ws_manager.run()：

   async def run(self):
       try:
           while True:
               await self.handle_ws_messages()  # 异步处理WS消息
               await asyncio.sleep(0.01)  # 主动让出CPU，避免独占事件循环
       except asyncio.CancelledError:
           # 在取消时执行优雅关闭逻辑
           await self.close_all_connections()
           raise  # 重新抛出异常，让上层任务处理流程完成
   

   如果任务中有同步代码（如同步DB查询），用asyncio.to_thread()包装：

   # 替换同步操作
   result = await asyncio.to_thread(sync_db_operation)
   

2. 修正lifespan中的任务关闭逻辑
   用asyncio标准的任务取消流程替代自定义的do_graceful_shutdown()调用，确保任务能正确响应取消信号：

   @asynccontextmanager
   async def lifespan(app: FastAPI):
       from .session import session
       from .sm import state_manager
       from .ws import ws_manager
   
       # 创建后台任务
       session_ping_task = asyncio.create_task(session.periodic_ping(60))
       ws_task = asyncio.create_task(ws_manager.run())
       state_task = asyncio.create_task(state_manager.try_restore_state())
   
       yield
   
       # 优雅取消所有任务
       for task in [ws_task, state_task, session_ping_task]:
           if not task.done():
               task.cancel()
               try:
                   await task  # 等待任务处理取消逻辑
               except asyncio.CancelledError:
                   pass
       
       print("完成优雅关闭")
   

问题2：BackgroundTasks导致接口等待任务完成 #

核心原因

• 任务中存在同步阻塞代码，阻塞了事件循环，导致FastAPI无法及时返回响应
• 对BackgroundTasks的工作机制误解：它会在请求响应返回后执行，但如果事件循环被占满，响应返回会被延迟

解决步骤

1. 替换BackgroundTasks为直接创建asyncio任务
   直接用asyncio.create_task()启动后台任务，确保任务立即进入事件循环调度，接口可以直接返回：

   @router.post("/update")
   async def update(new_val):
       logger.debug({'event': 'endpoint_called', 'endpoint': '/update'})
   
       # 直接创建后台任务，无需等待
       asyncio.create_task(start_update(new_val))
   
       return {'result': 'success'}
   

2. 为后台任务添加异常捕获
   直接创建的任务如果抛出异常不会被FastAPI捕获，需要在任务内部处理：

   async def start_update(new_val):
       try:
           a = await some_coro_1(new_val)
           b = await some_coro_2(a)
           # ... 其他异步步骤
           z = await some_coro_n(y)
       except Exception as e:
           logger.error(f"更新任务执行失败: {str(e)}")
   

3. 清理任务中的同步代码
   如果some_coro_x是同步函数，用asyncio.to_thread()包装成异步调用：

   async def some_coro_1(new_val):
       return await asyncio.to_thread(sync_some_coro_1, new_val)
   

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