Python中如何捕获程序所有异常？含asyncio场景捕获失败问题

问题根源 #

你的try-except无法捕获异常，通常是两个原因：

1. 第三方包内部启动了独立的asyncio子任务，这些任务的异常不在await MySomething的调用链中，外层try-except无法覆盖；
2. 你用loop.create_task()创建的main任务，其未捕获异常默认只会输出到控制台，不会被上层捕获。

具体解决方案 #

1. 设置全局事件循环异常处理器 #

给asyncio事件循环绑定自定义处理器，捕获所有未被任务自身处理的异常：

def custom_exception_handler(loop, context):
    print(f"未处理异常: {context['message']}")
    if 'exception' in context:
        print(f"异常详情: {context['exception']}")
    # 可在此添加日志写入、告警等扩展逻辑

loop = asyncio.new_event_loop()
loop.set_exception_handler(custom_exception_handler)
loop.create_task(main())
loop.run_forever()

2. 显式监控主任务的异常 #

不要直接用create_task启动main，而是用包裹函数捕获其异常：

async def run_main():
    try:
        await main()
    except Exception as e:
        print(f"主任务执行失败: {e}")

loop = asyncio.new_event_loop()
# 若需要持续运行，改用create_task后再启动循环
task = loop.create_task(run_main())
loop.run_forever()

3. 确保覆盖第三方包的连接逻辑 #

如果MySomething实例化后还需要等待连接完成或持续运行，要await其提供的对应方法（比如部分SocketIO客户端会有wait()或connect()协程）：

try:
    something = await MySomething("client", server)
    # 假设第三方包提供等待连接/运行的方法
    await something.wait()
except Exception as e:
    print(f"无法连接服务器 {server}: {e}")
    await asyncio.sleep(2)

补充：修复无限循环问题 #

你的代码中status变量从未被设置为"Connected"，会导致无限遍历服务器列表，需在连接成功后更新状态：

try:
    something = await MySomething("client", server)
    status = "Connected"
    break  # 退出服务器尝试循环
except Exception as e:
    print(f"无法连接服务器 {server}: {e}")
    await asyncio.sleep(2)

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