子线程异常无法被主线程直接捕获，需在子线程内处理或通过队列、Future等机制传递异常信息。1. 每个线程独立运行，未捕获的异常仅终止该线程；2. 可使用queue.Queue将异常传回主线程；3. 推荐concurrent.futures模块，其Future.result()会重新抛出异常；4. 自定义threading.excepthook（Python 3.8+）可统一记录线程异常；5. 必须主动收集异常，避免静默失败。
在多线程环境下使用 Python 的异常处理时，必须格外小心，因为主线程无法直接捕获子线程中抛出的异常。每个线程是独立执行的，未捕获的异常只会导致该线程终止，而不会影响主线程，这容易造成错误被忽略。
子线程异常不会传播到主线程
Python 中每个线程运行在独立的调用栈上，主线程 try-except 无法捕获子线程内的异常：
- 子线程中发生异常且未处理时，该线程会静默退出
- 主线程继续运行，可能误以为任务已完成
- 调试困难，日志缺失，问题难以追踪
通过共享变量或队列传递异常信息
为了在主线程感知子线程的异常状态，可以借助共享结构传递错误信息：
- 使用
queue.Queue将异常对象从子线程发送回主线程 - 定义一个结果容器（如字典或命名空间），在线程函数中记录异常
- 主线程等待线程完成后再检查结果是否包含异常
示例：
import threading import queuedef worker(q): try:
模拟任务
result = 1 / 0 q.put(('success', result)) except Exception as e: q.put(('error', e))q = queue.Queue() t = threading.Thread(target=worker, args=(q,)) t.start() t.join()
status, value = q.get() if status == 'error': print(f"子线程出错: {value}")
使用 concurrent.futures 更安全地管理异常
推荐使用
concurrent.futures模块替代原始 threading，它能自动封装异常并提供统一接口：
-
Future对象的result()方法会重新抛出异常 - 主线程可安全捕获并处理子线程异常
- 代码更简洁，减少手动管理线程的复杂性
示例：
from concurrent.futures import ThreadPoolExecutor import timedef task(): time.sleep(1) raise ValueError("出错了")
with ThreadPoolExecutor() as executor: future = executor.submit(task) try: future.result() except ValueError as e: print(f"捕获到子线程异常: {e}")
注意全局异常钩子和日志记录
为避免遗漏异常，可设置线程级别的异常钩子：
- 使用
sys.excepthook不适用于线程，应使用threading.excepthook（Python 3.8+） - 自定义
threading.excepthook来统一记录未捕获的线程异常 - 确保所有线程都配置了基本的日志输出，便于排查问题
示例：
import threading import sysdef custom_excepthook(args): print(f"线程异常: {args.exc_type.name}: {args.exc_value}")
设
置线程异常钩子
threading.excepthook = custom_excepthook
def bad_task(): raise RuntimeError("测试异常")
t = threading.Thread(target=bad_task) t.start() t.join() # 触发 custom_excepthook
置线程异常钩子基本上就这些。只要记得异常不会跨线程传播，主动收集和上报错误，就能有效避免多线程中“静默失败”的陷阱。
