如何捕获仅含asyncio栈跟踪的OSError?(Uvicorn+FastAPI环境)
Windows下FastAPI+Uvicorn异步API服务器的未捕获asyncio网络错误处理问题
我在Windows系统上用FastAPI和Uvicorn搭建了运行异步函数的API服务器。服务器偶尔会因asyncio抛出网络错误导致无响应,但我的脚本没法捕获这类错误。我想通过try-catch捕获错误后关闭并重启服务器,但错误回溯里完全没有我的代码。
项目文件结构
- API定义文件
app.py:
from fastapi import FastAPI app = FastAPI() @app.get("/check-status") async def root(): return {"Alive": True}
- 服务器配置文件
config.json:
{ "host": "0.0.0.0", "port": 80 }
- 启动脚本
run_app.py(已包含错误捕获逻辑):
import uvicorn import json from datetime import datetime if __name__ == '__main__': with open('path/to/config.json', 'r') as f: config = uvicorn.Config(**json.load(f)) server = uvicorn.Server(config) try: server.run() except Exception as e: err_timestamp = datetime.now().ctime() print(f'{err_timestamp}: Exception raised\n{e}\nRestarting...') server.shutdown() server.run()
触发的错误信息
服务器仍会因以下错误崩溃:
Task exception was never retrieved future: <Task finished name='Task-33094' coro=<IocpProactor.accept.<locals>.accept_coro() done, defined at C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py:563> exception=OSError(22, 'The specified network name is no longer available', None, 64, None)> Traceback (most recent call last): File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 566, in accept_coro await future File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 812, in _poll value = callback(transferred, key, ov) File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 555, in finish_accept ov.getresult() OSError: [WinError 64] The specified network name is no longer available Accept failed on a socket socket: <asyncio.TransportSocket fd=732, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('0.0.0.0', 80)> Traceback (most recent call last): File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 818, in loop conn, addr = f.result() File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 566, in accept_coro await future File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 812, in _poll value = callback(transferred, key, ov) File "C:\Users\Administrateur\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 555, in finish_accept ov.getresult() OSError: [WinError 64] The specified network name is no longer available
这个错误完全发生在asyncio内部,没有经过我的代码抛出,也没法复现测试。已知这类WinError是网络问题导致的,需要实现错误捕获来处理这种情况。
环境信息
- Python 3.8.10
- Uvicorn 0.17.4
- FastAPI 0.73.0
- Windows 10
解决方案
1. 捕获asyncio未处理的任务异常
这类错误属于异步任务未捕获的异常,不会直接向上抛出到server.run()的调用栈,常规try-catch无法捕获。可以给asyncio事件循环设置全局异常处理器,捕获这类未处理异常:
修改run_app.py如下:
import uvicorn import json import asyncio from datetime import datetime def handle_uncaught_exception(loop, context): # 打印错误日志 err_timestamp = datetime.now().ctime() print(f"{err_timestamp}: 捕获到未处理异常") print(f"异常详情: {context.get('exception', context['message'])}") # 异步触发服务器重启 asyncio.create_task(restart_server()) async def restart_server(): global server print("正在关闭服务器...") server.shutdown() print("正在重启服务器...") # 重新创建Server实例,避免复用旧实例的状态问题 with open('path/to/config.json', 'r') as f: config = uvicorn.Config(**json.load(f)) server = uvicorn.Server(config) await server.serve() if __name__ == '__main__': # 获取或创建事件循环 loop = asyncio.get_event_loop() # 设置全局未捕获异常处理器 loop.set_exception_handler(handle_uncaught_exception) with open('path/to/config.json', 'r') as f: config = uvicorn.Config(**json.load(f)) server = uvicorn.Server(config) try: loop.run_until_complete(server.serve()) except Exception as e: err_timestamp = datetime.now().ctime() print(f'{err_timestamp}: 捕获到异常\n{e}\n正在重启...') loop.run_until_complete(restart_server())
2. 升级依赖版本
当前使用的Uvicorn版本(0.17.4)较旧,后续版本对Windows下的asyncio异常处理有优化。可尝试升级到稳定新版本:
pip install --upgrade uvicorn fastapi
3. 线程池监控启动(备选方案)
如果全局异常处理仍不生效,可以用线程启动服务器,监控线程状态,当线程退出时自动重启:
import uvicorn import json import threading from datetime import datetime import time def start_server(): with open('path/to/config.json', 'r') as f: config = uvicorn.Config(**json.load(f)) server = uvicorn.Server(config) server.run() if __name__ == '__main__': while True: try: err_timestamp = datetime.now().ctime() print(f"{err_timestamp}: 启动服务器...") server_thread = threading.Thread(target=start_server) server_thread.start() server_thread.join() except Exception as e: err_timestamp = datetime.now().ctime() print(f'{err_timestamp}: 服务器异常退出\n{e}\n10秒后重启...') time.sleep(10)
内容的提问来源于stack exchange,提问作者Aidanpm
相关产品推荐
相关产品推荐

