子线程异常不会被主线程捕获，因为线程有独立执行栈，异常无法跨线程传播；可通过queue传递异常、封装函数返回结果与异常，或使用concurrent.futures模块在调用result()时重新抛出异常，实现可靠异常处理。
在Python多线程编程中，主线程无法直接捕获子线程中的异常。这意味着如果子线程发生错误，程序可能看似正常运行，但部分功能已经失效。要实现可靠的多线程程序，必须掌握异常的捕获与传递技巧。
为什么子线程异常不会被主线程捕获
每个线程拥有独立的执行栈，线程内未捕获的异常只会终止该线程，而不会传播到主线程。例如：
import threadingdef faulty_task(): raise ValueError("出错了！")
t = threading.Thread(target=faulty_task) t.start() t.join() # 即使join，也不会抛出异常 print("程序继续运行") # 这行仍会执行
上面代码中，虽然子线程抛出异常，但主线程完全不知情。
使用queue.Queue传递异常信息
通过共享队列将异常从子线程传回主线程，是最常见且可靠的方法。
import threading import queuedef worker(result_queue): try:
模拟任务
1 / 0 except Exception as e: result_queue.put(e) else: result_queue.put(None) # 成功完成result_q = queue.Queue() t = threading.Thread(target=worker, args=(result_q,)) t.start() t.join()
主线程检查结果
exc = result_q.get() if exc: raise exc # 在主线程重新抛出
这种方法适用于单个任务或有限数量的线程，能精确控制异常处理时机。
封装函数捕获异常并返回结果
更实用的方式是让线程返回（result, exception）元组，统一处理成功与失败情况。
def safe_run(func, *args, **kwargs):
try:
return func(*args, **kwargs), None
except Exception as e:
return None, e
def my_task(x, y):
return x / y
使用示例
result_q = queue.Queue()
def thread_wrapper():
result_q.put(safe_run(my_task, 5, 0))
t = threading.Thread(target=thread_wrapper)
t.start()
t.join()
result, exc = result_q.get()
if exc:
print(f"任务失败：{exc}")
else:
print(f"结果：{result}")
这种模式便于批量管理多个线程任务，清晰区分正常返回值和异常状态。
使用concurrent.futures简化异常处理
推荐方式：使用concurrent.futures模块，它自动封装异常，并在调用result()时重新抛出。
from concurrent.futures import ThreadPoolExecutor import timedef bad_task(): time.sleep(1) raise
RuntimeError("任务崩溃")
with ThreadPoolExecutor() as executor: future = executor.submit(bad_task) try: future.result() # 此处会重新抛出异常 except RuntimeError as e: print(f"捕获异常：{e}")
RuntimeError("任务崩溃")这是最简洁、安全的方式，适合大多数场景。future对象会保存异常，直到你主动获取结果。
基本上就这些。关键是要意识到线程隔离带来的异常不可见问题，主动设计传递机制。优先使用concurrent.futures，避免重复造轮子。手动管理时，用队列或返回结构体把异常“带出来”，才能真正掌控程序健壮性。
