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