在使用Python进行多进程开发时，子进程如果出现未捕获的异常会直接崩溃，主进程默认只能获取到子进程的退出状态码，无法得知具体的错误原因和详细的堆栈信息，这会让问题排查变得十分困难。下面介绍几种可行的方法，让主进程能够拿到子进程崩溃时的详细错误内容。
使用multiprocessing.Process的异常捕获
multiprocessing模块的Process类在子进程运行出现异常时，会将异常信息存储起来，主进程可以通过join方法等待子进程结束后，获取对应的异常对象。我们可以在子进程的目标函数外层包裹异常捕获逻辑，把异常信息通过进程安全的队列传递给主进程。
import multiprocessing
import traceback
import sys
def sub_process_task():
# 模拟子进程发生异常
result = 1 / 0
return result
def wrapper_task(queue):
try:
sub_process_task()
except Exception as e:
# 捕获异常，将异常类型和堆栈信息放入队列
error_info = {
"error_type": type(e).__name__,
"error_msg": str(e),
"traceback": traceback.format_exc()
}
queue.put(error_info)
# 重新抛出异常让子进程正常退出
raise
if __name__ == "__main__":
# 创建进程间通信队列
error_queue = multiprocessing.Queue()
# 创建子进程，目标函数为包装后的函数
process = multiprocessing.Process(target=wrapper_task, args=(error_queue,))
process.start()
process.join()
# 判断子进程是否异常退出
if process.exitcode != 0:
if not error_queue.empty():
error_detail = error_queue.get()
print("子进程崩溃详细信息：")
print(f"异常类型：{error_detail['error_type']}")
print(f"异常描述：{error_detail['error_msg']}")
print("完整堆栈：")
print(error_detail['traceback'])
else:
print(f"子进程退出码：{process.exitcode}，未获取到详细错误信息")
使用multiprocessing.Pool的错误回调
如果使用multiprocessing.Pool来管理进程池，可以通过注册错误回调函数来捕获子任务的异常信息。当子任务抛出异常时，错误回调会被触发，主进程可以在回调中拿到完整的异常对象。
import multiprocessing
import traceback
def pool_task(x):
# 模拟子任务异常
if x == 0:
raise ValueError("输入参数不能为0")
return 10 / x
def error_callback(error):
# 错误回调，接收子任务的异常对象
print("子任务执行出错，详细信息：")
print(f"异常类型：{type(error).__name__}")
print(f"异常描述：{str(error)}")
print("完整堆栈：")
traceback.print_exception(type(error), error, error.__traceback__)
if __name__ == "__main__":
with multiprocessing.Pool(processes=2) as pool:
# 提交任务时绑定错误回调
result = pool.apply_async(pool_task, args=(0,), error_callback=error_callback)
pool.close()
pool.join()
try:
# 尝试获取结果，正常情况会抛出异常
print(result.get())
except Exception as e:
print("主进程获取结果时捕获到异常：", str(e))
自定义进程类重写run方法
我们还可以通过继承multiprocessing.Process类，重写run方法，在run方法中添加全局的异常捕获逻辑，将错误信息存储到实例属性中，主进程等待子进程结束后可以直接读取这些属性。
import multiprocessing
import traceback
class CustomProcess(multiprocessing.Process):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.error_info = None
def run(self):
try:
# 调用父类默认的run逻辑，执行target函数
super().run()
except Exception as e:
# 捕获所有异常，存储详细信息
self.error_info = {
"error_type": type(e).__name__,
"error_msg": str(e),
"traceback": traceback.format_exc()
}
# 可以选择将异常重新抛出，也可以直接处理
raise
def sub_task():
# 模拟子进程异常
my_list = [1, 2, 3]
print(my_list[5])
if __name__ == "__main__":
process = CustomProcess(target=sub_task)
process.start()
process.join()
if process.exitcode != 0:
if process.error_info:
print("子进程崩溃详细信息：")
print(f"异常类型：{process.error_info['error_type']}")
print(f"异常描述：{process.error_info['error_msg']}")
print("完整堆栈：")
print(process.error_info['traceback'])
else:
print(f"子进程退出码：{process.exitcode}")
方案对比
不同方案的适用场景和特点如下：
| 方案 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| Process+队列传递 | 单独管理子进程的场景 | 逻辑清晰，错误信息传递稳定 | 需要额外创建队列，代码稍多 |
| Pool错误回调 | 进程池批量处理任务的场景 | 和进程池结合紧密，使用简单 | 仅适用于进程池场景 |
| 自定义进程类 | 需要复用进程逻辑的场景 | 可扩展性强，封装性好 | 需要自定义类，理解成本稍高 |
注意事项
- 所有的异常信息捕获需要在子进程内部完成，因为子进程的异常不会自动传递到主进程。
- 使用队列传递错误信息时，要选择合适的队列类型，multiprocessing.Queue是进程安全的，适合该场景。
- 如果子进程是被外部信号杀死而不是异常退出，上述方案无法捕获到错误堆栈，此时只能通过退出码判断异常退出。
- 在多进程场景下，traceback.format_exc()生成的堆栈信息是子进程内的调用栈，和主进程的调用栈是独立的，排查问题时需要注意。
Python子进程主进程错误捕获multiprocessing修改时间：2026-06-18 19:33:43