本文案例来自 psf/requests GitHub Issue #6628，适配至国内生产环境后验证。
TL;DR
用 concurrent.futures.ProcessPoolExecutor 并发调 API，某个 worker 返回了 requests.exceptions.JSONDecodeError，结果不是那一个 task 失败——整个进程池崩溃，所有正在跑的 task 全部丢失。根因是 JSONDecodeError 实例的 pickle 序列化在某些条件下会失败，导致 worker 进程异常终止 → BrokenProcessPool。解决方案：方案 A（包装异常为可序列化类型）、方案 B（改用 ThreadPoolExecutor）、方案 C（json.loads() 预检）。
现象
环境
from concurrent.futures import ProcessPoolExecutor, as_completed
import requests
def fetch_json(url):
resp = requests.get(url)
return resp.json() # ← 这里可能抛出 JSONDecodeError
with ProcessPoolExecutor(max_workers=4) as executor:
futures = {executor.submit(fetch_json, url): url for url in urls}
for future in as_completed(futures):
try:
result = future.result()
except Exception as e:
print(f"Task failed: {e}")
完整报错
concurrent.futures.process.BrokenProcessPool:
A process in the process pool was terminated abruptly
while the future was running or pending.
直觉告诉你：capture 了 Exception，一个 task 失败不应该影响其他 task。但实际：所有正在跑的 task 全部超时/shutdown，进程池被销毁重建。
排查过程
第一回合：只看异常类型
你捕获的是 Exception，但 BrokenProcessPool 不是从 JSONDecodeError 传播来的——它是 ProcessPoolExecutor 检测到 worker 进程非正常退出后抛出的。
# BrokenProcessPool 不是 JSONDecodeError 的父类
# ——它是一个信号：worker 进程死了
第二回合：模拟最小复现
import pickle
from requests.exceptions import JSONDecodeError
# 模拟 API 返回非 JSON
import json
try:
json.loads("not json{")
except json.JSONDecodeError as e:
err = JSONDecodeError("msg", "doc", 0)
# 尝试序列化它
try:
pickle.dumps(err)
except Exception as e2:
print(f"Pickle failed: {e2}")
输出：
Pickle failed: Can't pickle ...: it's not the same object as ...
找到了——JSONDecodeError 不能 pickle。
第三回合：为什么 worker 进程会死
Python 多进程通信流程：
Worker 子进程
↓ 执行 task，返回 result 或 exception
↓ pickle.dumps(exception) ← 这里炸了
↓ 子进程因未捕获异常而退出（exit code != 0）
主进程
↓ 检测到 worker pid 消失
↓ 抛出 BrokenProcessPool
JSONDecodeError.__reduce__() 在特定条件下（如通过 requests 包装后）会触发 pickle.PicklingError，子进程的序列化线程崩溃 → 进程退出 → 主进程看到 worker 死了 → BrokenProcessPool。
关键点：不是 JSON 解析失败导致崩溃——是「把失败结果传回主进程」这个动作导致了崩溃。
根因
Python pickle 对异常对象的序列化限制
Python 3 的异常对象 __reduce__ 默认行为：
# 异常默认 pickle 方式：用 __init__ 参数重建
def __reduce__(self):
return (type(self), self.args)
但 requests.exceptions.JSONDecodeError 的 args 包含了不可序列化的对象引用（例如底层 urllib3 的 response 对象引用链），导致 pickle 失败。
而 json.JSONDecodeError（标准库）本身是可 pickle的——问题出在 requests 的包装层：
# requests/exceptions.py
class JSONDecodeError(RequestException, json.JSONDecodeError):
def __init__(self, msg, doc, pos):
# doc 是原始响应体 bytes/str，可能导致 pickle 引用链问题
三层因果链
1. API 返回非 JSON 内容
↓
2. resp.json() 抛出 requests.exceptions.JSONDecodeError
↓
3. ProcessPoolExecutor 尝试 pickle.dumps(该异常) 传回主进程
↓
4. pickle 失败 → worker 子进程 crash → BrokenProcessPool
解决方案（按推荐度排列）
方案 A：包装异常（推荐 ✅）
在 worker 函数内部捕获所有异常，包装为可序列化类型：
class TaskError(Exception):
"""可序列化的任务异常包装"""
def __init__(self, error_type: str, error_msg: str):
self.error_type = error_type
self.error_msg = error_msg
super().__init__(error_msg)
def __reduce__(self):
return (TaskError, (self.error_type, self.error_msg))
def fetch_json_safe(url):
try:
resp = requests.get(url, timeout=10)
return resp.json()
except Exception as e:
raise TaskError(type(e).__name__, str(e)) from None
为什么 from None：切断异常链，避免 context 引入不可序列化对象。
方案 B：换 ThreadPoolExecutor（适合 IO 密集型）
如果你的 task 主要是网络 IO（调 API、读写文件），ThreadPoolExecutor 几乎没有 GIL 开销，且不涉及 pickle：
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=8) as executor:
# 完全相同的接口，线程间共享内存，无需 pickle
选 B 的判断标准：task 中 CPU 密集操作 < 10%，网络/磁盘 IO > 90% → 直接用线程池。
方案 C：json.loads() 预检（适合已知 API 格式）
def fetch_json_safe(url):
resp = requests.get(url, timeout=10)
text = resp.text
# 预检：尝试解析，失败则返回 None + 日志
try:
return json.loads(text)
except json.JSONDecodeError:
# 标准库 json.JSONDecodeError 是可 pickle 的！
raise # 这个可以从进程池正常传回
为什么标准库的 json.JSONDecodeError 可行：它是纯 Python 对象，args 只含 (msg, doc, pos) 三个基础类型，pickle 无压力。
方案 D：设置进程池初始器
import signal
def worker_init():
signal.signal(signal.SIGTERM, signal.SIG_DFL)
with ProcessPoolExecutor(max_workers=4, initializer=worker_init) as executor:
...
只缓解不根治——让 worker 在被 kill 时优雅退出，但不能阻止 pickle 失败导致的 crash。
如何自检你是否中招
# 快速测试：你的异常对象能不能 pickle？
import pickle
from requests.exceptions import JSONDecodeError
err = JSONDecodeError("test", "{}", 0)
try:
pickle.dumps(err)
print("✅ 可序列化，安全")
except Exception as e:
print(f"❌ 不可序列化: {e}")
print("→ 不要从 ProcessPoolExecutor 内抛出此异常")
启示
ProcessPoolExecutor不是ThreadPoolExecutor的「更高级版本」。 多进程 = pickle 序列化边界 = 所有返回值和异常都必须可 pickle。- 第三方库包装的异常类可能不可序列化。
requests.JSONDecodeError继承了两个父类，__reduce__行为可能不同于标准库。 BrokenProcessPool的意思是「有 worker 死了」，不是「你的 task 失败了」。 先查 worker 退出原因，不要直接重试。- IO 密集型用线程池，CPU 密集型用进程池。 「调 API」是 IO 密集——用
ThreadPoolExecutor即避开了 pickle 问题，也省去了进程 fork 开销。 - 在 worker 函数中加一层 try/except + 自定义可序列化异常包装——这是防御多进程序列化问题的最短路径。
原始出处： - psf/requests #6628: JSONDecodeError can't be deserialized - crashes BrokerProcessPool —— mikelane 首次报告并提出修复 PR - Python pickle 官方文档：
__reduce__协议 - CPythonconcurrent.futures.process源码 —— worker 退出检测逻辑在_queue_management_thread
本文首发于 CSDN 专栏《Python 生产环境报错速查：从崩溃到修复》，转载请注明出处。
本文由 admin 原创，转载请注明出处。
评论
0