Python multiprocessing 进程池传递不可序列化对象

我第一次遇到这个问题，是把一个本地能跑的批处理改成进程池后，任务一提交就报 Can't pickle local object。后来换到另一台机器，错误又变成参数不能序列化，表面像是平台差异，实际是进程池终于把对象边界暴露出来了。multiprocessing 的任务函数、参数和返回值都要经过跨进程传递，不能把线程里的“共享内存直觉”直接搬过来。

官方文档：https://docs.python.org/3/library/multiprocessing.html

最稳妥的修复不是给任意对象强行加序列化钩子，而是让进程池只接收模块顶层函数和简单数据描述；文件、连接、锁等运行时资源在子进程内部初始化。这样代码才不依赖某台机器恰好使用 fork。

我第一次遇到的不是 Pool 坏了

进程池失败的位置通常有三处：父进程把任务函数和参数送给 worker 时，worker 把结果送回父进程时，以及回调函数或异常对象需要跨进程返回时。只要其中一处包含局部函数、lambda、打开的文件、线程锁或不能恢复的第三方运行时对象，就可能出现 PicklingError、AttributeError 或更晚才出现的结果读取异常。

Python 官方对 pickle 的边界很明确：模块顶层定义的函数和类通常可以按限定名定位，函数体本身不会被打包；包含不可序列化成员的容器或实例仍然会失败。因而“把函数移到外面”只能解决函数位置问题，不能让函数闭包里捕获的连接和文件句柄突然变得可传递。

先把进程池的传值边界拆开

排查时我会先把调用改写成一个极小的任务协议：函数只描述计算，参数只保存数据，返回值只保存结果。下面这个写法把任务函数放在模块顶层，并在入口保护内创建进程池，适合 spawn 和 forkserver 都参与的环境。

from multiprocessing import get_context

def normalize_record(record):
    # 任务函数在模块顶层，子进程可以按模块名重新找到它
    return {"name": record["name"].strip(), "score": int(record["score"])}

if __name__ == "__main__":
    # 显式选择上下文，避免把机器默认值当成业务协议
    ctx = get_context("spawn")
    records = [
        {"name": " Alice ", "score": "8"},
        {"name": " Bob ", "score": "9"},
    ]
    with ctx.Pool(processes=2) as pool:
        # 参数是由基础类型组成的数据描述，结果也保持为普通字典
        results = pool.map(normalize_record, records)
    print(results)

这里的关键不是 spawn 一定比其他方式更快，而是用它主动暴露“能否独立导入和重建”的问题。pool.map 的输入不要放连接对象、打开的文件或带线程状态的实例；需要传递的配置可以先变成字符串、数字、列表、字典或一个能正常重建的顶层类实例。

Python 3.14 之后，旧代码的“能跑”不再是保证

这次迁移最容易忽略的变化是启动方式。官方文档记录了：Windows 和 macOS 默认使用 spawn；从 Python 3.14 起，POSIX 平台不再默认使用 fork，支持的平台默认转为 forkserver。依赖父进程内存状态的旧代码，升级后可能第一次遇到局部函数、全局变量初始化顺序和运行时资源不能传递的问题。

fork 会让子进程继承父进程的大量状态，这会掩盖对象没有明确初始化的问题；spawn 会启动新的解释器，forkserver 则从专用服务进程创建子进程。三者的可用平台和资源继承规则不同。业务代码不应靠“默认就是 fork”来约定行为，库代码尤其应该允许调用方提供自己的 multiprocessing context。

import multiprocessing as mp

def run_job(value):
    # 只消费可描述的值，避免依赖父进程中的隐式全局状态
    return value * value

def main():
    # 把启动方式作为部署配置的一部分记录和回归
    ctx = mp.get_context("forkserver")
    with ctx.Pool(2) as pool:
        return pool.map(run_job, [1, 2, 3])

if __name__ == "__main__":
    # spawn/forkserver 需要安全导入主模块，不能在导入阶段创建 Pool
    print(main())

如果项目确实要求 fork，就显式调用 get_context("fork") 或在受保护入口中设置启动方式，并把这个选择写进部署文档。这样以后看到失败时，能区分是对象协议改变，还是启动方式被环境改变，而不是靠重试碰运气。

可序列化重构怎么落地

我最后保留了四条简单规则。第一，任务函数放到模块顶层，避免局部函数和 lambda。第二，把复杂对象转换为数据传输对象，只传必要字段。第三，数据库连接、文件句柄、客户端和模型实例等资源，在 worker 内通过 initializer 或任务函数第一次调用时创建。第四，返回值尽量是小而明确的记录，避免把整个运行时对象送回父进程。

from multiprocessing import get_context

worker_client = None

def init_worker(endpoint):
    global worker_client
    # 连接属于子进程资源，在 worker 内创建，不跨进程传递句柄
    worker_client = create_client(endpoint)

def fetch_one(item):
    # 入参只保留可序列化的标识，返回值只保留业务结果
    data = worker_client.fetch(item["key"])
    return {"key": item["key"], "size": len(data)}

if __name__ == "__main__":
    # 这里的工厂函数和配置字符串都能被 spawn 重新导入与传递
    ctx = get_context("spawn")
    with ctx.Pool(2, initializer=init_worker, initargs=("service.local",)) as pool:
        rows = pool.map(fetch_one, [{"key": "a"}, {"key": "b"}])
    print(rows)

示例中的 create_client 代表项目自己的客户端工厂，不能把已经打开的客户端实例塞进 initargs。此外，pickle 本身不适合处理不可信输入；这里只讨论同一程序内部为进程池传递受控对象，不要把它理解成可以安全反序列化外部数据的通用方案。

我的迁移回归清单

最终判断标准不是某一台开发机上能否得到结果，而是换成项目声明的启动方式后，任务函数、输入、输出和资源生命周期仍然清楚。这样处理后，进程池的错误会从“偶发平台问题”变成能定位到具体传值边界的工程问题。

相关问题

为什么顶层函数可以传，lambda 不行？

进程间传递时，顶层函数通常能通过模块和限定名重新定位；lambda 没有稳定的可导入名称，闭包还可能捕获额外状态。把逻辑写成模块顶层的具名函数更稳妥。

ProcessPoolExecutor 也有同样的问题吗？

有。它使用进程池把任务提交到后台进程，函数、参数和结果同样受可序列化边界影响。若需要更高层的提交接口，可以换 API，但不能绕过对象传递规则。