Python的threading模块创建的线程中，如果运行逻辑抛出异常，默认不会向上传递到创建它的主线程或者父线程，这会导致异常被静默忽略，程序可能出现不符合预期的运行结果，因此掌握线程异常的捕获和处理方法非常重要。
为什么线程异常默认无法被外部捕获
Python的线程是独立运行的执行单元，线程的调用栈和主线程是相互隔离的。当线程内部的函数抛出异常时，异常只会在当前线程的调用栈中传播，如果没有在线程内部处理，线程会直接终止，异常不会自动传递到主线程的调用栈中，所以主线程中用常规的try...except块无法捕获到线程内部的异常。
常见的线程异常捕获方案
方案一：重写Thread类，自定义异常存储逻辑
我们可以通过继承threading.Thread类，重写run方法，在线程内部捕获异常并存储到实例属性中，之后在主线程中读取该属性来判断线程是否出现异常。
import threading
class ExceptionThread(threading.Thread):
def __init__(self, target=None, args=(), kwargs=None):
super().__init__()
self.target = target
self.args = args
self.kwargs = kwargs if kwargs is not None else {}
self.exception = None
def run(self):
try:
if self.target:
self.target(*self.args, **self.kwargs)
except Exception as e:
# 捕获线程内的所有异常，存储到实例属性中
self.exception = e
def task(num):
# 模拟线程内抛出异常的场景
if num < 0:
raise ValueError("参数不能为负数")
print(f"任务执行完成，参数为{num}")
# 创建线程实例
thread1 = ExceptionThread(target=task, args=(10,))
thread2 = ExceptionThread(target=task, args=(-5,))
thread1.start()
thread2.start()
thread1.join()
thread2.join()
# 主线程中检查线程是否有异常
if thread1.exception:
print(f"线程1出现异常：{thread1.exception}")
else:
print("线程1执行正常")
if thread2.exception:
print(f"线程2出现异常：{thread2.exception}")
else:
print("线程2执行正常")
方案二：使用队列传递异常信息
队列是线程安全的数据结构，我们可以在线程内部捕获异常后，将异常对象或者异常信息放入队列，主线程从队列中读取来判断各个线程的执行情况。
import threading
import queue
def task_with_queue(num, error_queue):
try:
if num < 0:
raise ValueError("参数不能为负数")
print(f"任务执行完成，参数为{num}")
except Exception as e:
# 将异常信息放入队列
error_queue.put((threading.current_thread().name, e))
# 创建线程安全的队列
error_queue = queue.Queue()
threads = []
# 创建两个线程，一个正常执行，一个抛出异常
for num in [10, -5]:
thread = threading.Thread(target=task_with_queue, args=(num, error_queue))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
# 主线程从队列中读取异常信息
while not error_queue.empty():
thread_name, exception = error_queue.get()
print(f"线程{thread_name}出现异常：{exception}")
方案三：设置全局异常钩子捕获线程异常
Python的threading模块提供了excepthook属性，我们可以自定义异常钩子函数，当线程内出现未捕获的异常时，会调用这个钩子函数，我们可以在钩子函数中统一处理所有线程的异常。
import threading
import sys
def thread_exception_handler(args):
# args包含异常信息、线程对象等
thread = args.thread
exception = args.exc_value
print(f"线程{thread.name}出现未捕获异常：{exception}")
# 设置全局线程异常钩子
threading.excepthook = thread_exception_handler
def task(num):
if num < 0:
raise ValueError("参数不能为负数")
print(f"任务执行完成，参数为{num}")
thread1 = threading.Thread(target=task, args=(10,), name="正常线程")
thread2 = threading.Thread(target=task, args=(-5,), name="异常线程")
thread1.start()
thread2.start()
thread1.join()
thread2.join()
不同方案的选择建议
如果只需要针对单个线程处理异常，重写Thread类的方案更轻量，代码逻辑也更清晰；如果需要在主线程中统一收集多个线程的异常信息，使用队列的方案更合适，方便批量处理；如果需要全局捕获所有线程的未处理异常，设置全局异常钩子的方案更高效，不需要修改每个线程的实现逻辑。
需要注意的是，无论使用哪种方案，都建议在线程内部尽可能处理可预期的异常，避免异常无限制扩散，同时结合日志记录异常信息，方便后续排查问题。