如何定位Python库中抛出asyncio.exceptions.CancelledError异常的代码行?
捕获完整异常栈
不要只捕获CancelledError就终止流程,必须打印完整调用栈才能定位根源。用traceback模块输出详细异常信息:import traceback import asyncio import logging logger = logging.getLogger(__name__) async def main(): tasks = [...] # 你的任务列表 try: await asyncio.gather(*tasks) except asyncio.CancelledError: traceback.print_exc() logger.error("asyncio.gather被取消", exc_info=True)异常栈会明确显示取消操作的发起位置——是Uvicorn内部代码,还是某个任务的调用链。
检查Uvicorn超时配置
Uvicorn的超时机制会主动取消超时请求任务,常见参数包括timeout_keep_alive、timeout_graceful_shutdown、timeout_notify;如果用FastAPI,还要注意路由级的timeout参数。
临时调高超时参数测试,比如启动命令:uvicorn main:app --host 0.0.0.0 --port 8000 --timeout-keep-alive 60 --timeout-graceful-shutdown 30或代码中配置:
if __name__ == "__main__": uvicorn.run( "main:app", host="0.0.0.0", port=8000, timeout_keep_alive=60, timeout_graceful_shutdown=30 )若调高后异常消失,说明是超时触发的任务取消。
逐个隔离任务定位问题
不要一次性运行所有任务,先单独运行每个任务,再逐步叠加,找到触发取消的具体任务:# 先测试单个任务 await asyncio.gather(tasks[0]) # 无异常再加第二个任务 await asyncio.gather(tasks[0], tasks[1])定位到问题任务后,单独调试其执行流程,检查是否内部有取消逻辑,或依赖的异步操作抛出了CancelledError。
给任务添加取消监控
给每个任务包装一层取消捕获,打印任务内的调用栈,明确哪个任务先被取消:async def monitored_task(task_func, *args, **kwargs): try: return await task_func(*args, **kwargs) except asyncio.CancelledError: print(f"任务 {task_func.__name__} 被取消") traceback.print_exc() raise # 重新抛出,保留原有取消流程 # 创建任务时使用包装器 tasks = [monitored_task(task1), monitored_task(task2)] await asyncio.gather(*tasks)检查外部关闭信号
Uvicorn收到SIGINT(Ctrl+C)、SIGTERM(进程终止)信号时,会触发优雅关闭并取消所有运行中的任务。通过生命周期钩子监控关闭事件:from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): print("服务启动") yield print("服务开始关闭,任务将被取消") app = FastAPI(lifespan=lifespan)若异常出现在关闭阶段,说明是正常优雅关闭流程;若不是预期关闭,排查是否有意外信号触发(如容器重启、进程误杀)。
临时禁用Uvicorn信号处理
启动时添加--no-sigint --no-sigterm参数,禁止Uvicorn处理中断信号,测试是否还会触发CancelledError:uvicorn main:app --no-sigint --no-sigterm若异常消失,说明是外部信号导致的任务取消。
排查第三方异步库
如果任务中使用了第三方异步库,部分库可能会主动取消任务或内部抛出CancelledError。将第三方库调用替换为模拟逻辑,逐步排查是否是依赖库的问题。
内容的提问来源于stack exchange,提问作者Shakir

