Python 如何捕获未处理的全局异常

sys

sys.excepthook

try...except

sys.excepthook

exc_type

exc_value

exc_traceback

sys.excepthook

import sys
import logging
import traceback

# 配置日志，以便将异常信息写入文件
logging.basicConfig(level=logging.ERROR,
                    filename='application_errors.log',
                    filemode='a', # 追加模式
                    format='%(asctime)s - %(levelname)s - %(message)s')

def custom_global_exception_handler(exc_type, exc_value, exc_traceback):
    """
    自定义的全局异常处理函数。
    它会捕获所有未被处理的异常，并进行日志记录。
    """
    # 避免捕获 KeyboardInterrupt，让它正常退出（例如用户按 Ctrl+C）
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return

    # 记录异常的详细信息到日志文件
    error_message = f"Unhandled exception caught!\n" \
                    f"Type: {exc_type.__name__}\n" \
                    f"Value: {exc_value}\n" \
                    f"Traceback:\n{''.join(traceback.format_tb(exc_traceback))}"
    logging.error(error_message)

    # 在控制台给用户一个友好的提示
    print("\n哎呀！程序遇到一个意想不到的错误，我们已经记录下来了。")
    print("请查看日志文件 'application_errors.log' 获取更多详情。")

    # 如果需要，你可以在这里执行一些清理工作，或者安全地退出程序
    # sys.exit(1)

# 将全局异常钩子设置为我们的自定义处理函数
sys.excepthook = custom_global_exception_handler

# 模拟一个会触发未处理异常的代码
def problematic_operation():
    print("正在执行一个可能出错的操作...")
    result = 1 / 0 # 这会引发 ZeroDivisionError
    return result

# problematic_operation() # 取消注释来测试

# 另一个例子：访问列表越界
# my_list = [1, 2, 3]
# print(my_list[5]) # 这会引发 IndexError

print("程序正常启动，等待异常发生...")
# 为了让程序持续运行，以便观察sys.excepthook的捕获效果，
# 可以在这里放置一些长时间运行的代码或等待用户输入
# input("按 Enter 键触发一个异常...")
# problematic_operation()

logging

custom_global_exception_handler

try-except

try-except

try-except

try-except

try-except

asyncio

sys.excepthook

sys.excepthook

try-except

sys.excepthook

threading.Thread

run

try-except

import threading
import time

def worker_function():
    print(f"线程 {threading.current_thread().name} 启动...")
    time.sleep(1)
    # 这个异常不会被 sys.excepthook 捕获
    result = 1 / 0
    print(f"线程 {threading.current_thread().name} 结束。")

def safe_worker_function():
    try:
        worker_function()
    except Exception as e:
        # 在子线程内部捕获并处理异常
        logging.error(f"线程 {threading.current_thread().name} 发生异常: {e}", exc_info=True)
        print(f"线程 {threading.current_thread().name} 内部捕获到异常。")

# thread = threading.Thread(target=worker_function, name="BadWorker")
# thread.start()
# thread.join() # 等待线程结束

safe_thread = threading.Thread(target=safe_worker_function, name="GoodWorker")
safe_thread.start()
safe_thread.join()

sys.excepthook

asyncio

asyncio

asyncio

sys.excepthook

asyncio

sys.stderr

loop.set_exception_handler()

context

asyncio

sys.excepthook

import asyncio
# ... logging setup from previous example ...

def custom_asyncio_exception_handler(loop, context):
    """
    自定义的 asyncio 事件循环异常处理函数。
    """
    exception = context.get("exception")
    if exception:
        logging.error(f"Unhandled asyncio exception: {exception}", exc_info=True)
    else:
        logging.error(f"Unhandled asyncio exception context: {context}")
    print("\n异步任务中发生一个错误，已记录。")
    # 这里可以根据情况选择停止循环，或者做其他清理
    # loop.stop()

async def faulty_async_task():
    print("异步任务开始...")
    await asyncio.sleep(0.5)
    raise ValueError("异步任务中出错了！")
    print("异步任务结束。") # 这行不会执行

async def main_async_app():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(custom_asyncio_exception_handler)

    task = asyncio.create_task(faulty_async_task())
    try:
        await task
    except ValueError as e:
        # 如果这里捕获了，就不会触发 loop.set_exception_handler
        print(f"主应用局部捕获到异步任务异常: {e}")
    await asyncio.sleep(1) # 保持事件循环运行一段时间

# asyncio.run(main_async_app())

sys.excepthook

try-except

asyncio

loop.set_exception_handler()

sys.excepthook

asyncio

custom_global_exception_handler