Flask中用multiprocessing.Pool遇无法序列化Environment对象错误

问题背景 #

你在Flask应用里尝试用multiprocessing.Pool做并发处理，通过蓝图调用服务类的方法，但一直碰到pickle错误——哪怕把业务逻辑简化到只打印参数，还是报TypeError: cannot pickle 'Environment' object，但同样的代码在Jupyter Notebook里却能正常运行。

先回顾下你的核心代码结构：

blueprint.py #

@bp.route(...)
def my_bp_route(...):
    service = get_service_instance()
    service.do_stuff(...)

service.py（简化前） #

import multiprocessing as mp

def foo(self, ...):
    # 执行操作

def do_stuff(self, ...):
    some_tuple_pairs = [
        (CustomType.TypeA.value, IdStr.TypeA.value),
        (CustomType.TypeB.value, IdStr.TypeB.value),
    ]
    cropbox = (5, 5)
    # LTChar来自pdfminer
    self.do_mp_concurrent(some_tuple_pairs, [LTChar(), LTChar(), ...], cropbox, [{'key1': 1, ...}, ...])

def do_mp_concurrent(
    self, types_to_check: List[Tuple[str, str]], pdf_objs: List[LTChar], cropbox: Tuple[int, int], some_dicts: List[dict]
):
    with mp.Pool(processes=3) as pool:
        results = pool.starmap(
            self.foo, [(cus_type, id_str, pdf_objs, cropbox, some_dicts) for cus_type, id_str in types_to_check]
        )
        ...

简化后仍报错的代码 #

def foo(self, one, two):
    print(one, two)

def do_mp_concurrent(
    self, types_to_check: List[Tuple[str, str]], pdf_objs: List[LTChar], cropbox: Tuple[int, int], some_dicts: List[dict]
):
    with mp.Pool(processes=3) as pool:
        results = pool.starmap(
            self.foo, [(1, 2)]
        )
        ...

错误栈 #

results = pool.starmap(
  File "/usr/lib64/python3.8/multiprocessing/pool.py", line 372, in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
  File "/usr/lib64/python3.8/multiprocessing/pool.py", line 771, in get
    raise self._value
  File "/usr/lib64/python3.8/multiprocessing/pool.py", line 537, in _handle_tasks
    put(task)
  File "/usr/lib64/python3.8/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/usr/lib64/python3.8/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: cannot pickle 'Environment' object

问题根源 #

问题出在Flask的请求上下文被绑定到了你的服务类实例上。当你把self.foo传给Pool.starmap时，Python需要把整个服务类实例（self）序列化（pickle）后传递给子进程——而Flask的Environment对象（属于Flask上下文的一部分）是无法被pickle的。

为什么Jupyter里能跑？因为Jupyter环境没有Flask的请求上下文，你的服务类实例里没有绑定那些不可pickle的Flask对象，所以序列化没问题。

解决方案 #

这里给你几个可行的解决思路，按推荐程度排序：

1. 把并发逻辑改成独立函数，避免传递类实例 #

最直接的办法是把foo从服务类里抽出来，变成独立函数，所有需要的参数都显式传递，不要依赖self里的Flask相关对象：

# service.py
import multiprocessing as mp

# 独立的foo函数，不需要self
def foo(cus_type, id_str, pdf_objs, cropbox, some_dicts):
    # 原来的foo逻辑，直接用传入的参数即可
    pass

class Service:
    def do_stuff(self, ...):
        # ... 其他代码 ...
        self.do_mp_concurrent(...)

    def do_mp_concurrent(self, types_to_check, pdf_objs, cropbox, some_dicts):
        with mp.Pool(processes=3) as pool:
            results = pool.starmap(
                foo, [(cus_type, id_str, pdf_objs, cropbox, some_dicts) for cus_type, id_str in types_to_check]
            )
            # 处理结果

2. 使用spawn上下文创建Pool #

Flask默认用fork方式启动子进程，会继承父进程的Flask上下文。改用spawn方式启动子进程，会重新初始化Python解释器，避免继承不可pickle的对象：

def do_mp_concurrent(self, types_to_check, pdf_objs, cropbox, some_dicts):
    # 获取spawn上下文
    ctx = mp.get_context('spawn')
    with ctx.Pool(processes=3) as pool:
        results = pool.starmap(
            self.foo, [(cus_type, id_str, pdf_objs, cropbox, some_dicts) for cus_type, id_str in types_to_check]
        )
        # ...

⚠️ 注意：用spawn方式时，要确保你的Flask应用启动代码放在if __name__ == '__main__':块里，避免子进程重复初始化Flask：

# app.py
from flask import Flask

def create_app():
    app = Flask(__name__)
    # 注册蓝图等配置
    return app

if __name__ == '__main__':
    app = create_app()
    app.run()

3. 改用concurrent.futures.ProcessPoolExecutor #

concurrent.futures的API更简洁，同样支持指定spawn启动方式，效果和上面的方法一致：

from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp

def do_mp_concurrent(self, types_to_check, pdf_objs, cropbox, some_dicts):
    with ProcessPoolExecutor(max_workers=3, mp_context=mp.get_context('spawn')) as executor:
        results = list(executor.starmap(
            self.foo, [(cus_type, id_str, pdf_objs, cropbox, some_dicts) for cus_type, id_str in types_to_check]
        ))
        # ...

4. 用任务队列替代直接启动子进程（推荐生产环境） #

如果是生产环境的Flask应用，不建议在请求处理过程中直接启动子进程——这会导致资源管理混乱，也容易引发上下文问题。更好的方式是用Celery、RQ这类任务队列，把耗时的并发任务放到后台执行：

• 把do_mp_concurrent里的逻辑改成发送任务到队列
• 启动独立的worker进程处理任务
• 请求只负责触发任务，不等待任务完成（如果需要结果，可以用异步回调或轮询）

内容的提问来源于stack exchange，提问作者bvbav