如何在 Python 多进程环境中安全使用 NumPy 向量化函数

本文详解为何 `np.vectorize()` 创建的函数无法被 `multiprocessing` 或 `pathos.multiprocess` 正确序列化，并提供可落地的解决方案：延迟初始化向量化函数，确保其在子进程中构建而非跨进程传递。

在使用 multiprocessing.Process 或 pathos.multiprocess 进行并行计算时，Python 依赖 pickle（或 dill）对函数、对象及其状态进行序列化与反序列化，以在子进程中重建执行环境。然而，np.vectorize() 返回的对象是一个特殊的 numpy.vectorize 实例，其底层封装了原始函数（如 _wind_dfn）并生成一个不可直接 pickle 的 ufunc-类对象。当该对象作为类属性在主进程初始化（如 __init__ 中赋值），再随实例传入子进程时，pickle 会尝试序列化该向量化函数——但因其动态生成、无全局可导入路径（不在 __main__ 或模块顶层命名空间中），最终抛出类似以下错误：

_pickle.PicklingError: Can't pickle : attribute lookup _wind_dfn (vectorized) on __main__ failed

根本原因：np.vectorize 不是“纯函数构造器”，它返回的实例绑定着闭包和内部状态，且 pickle 无法可靠还原其来源；而 multiprocessing 要求所有跨进程传递的对象必须可被 pickle 完整序列化。

✅ 推荐解决方案：惰性初始化（Lazy Initialization）
避免在 __init__ 中预先创建 np.vectorize 对象，改为在首次调用时按需构建，并缓存于实例属性中。这样，向量化函数总是在当前进程内创建，完全绕过序列化瓶颈。

以下是修正后的代码结构（基于原示例优化）：

import abc
import numpy as np
from multiprocessing import Process

class ProblemClassBase(metaclass=abc.ABCMeta):
    def __init__(self):
        # ✅ 不在此处创建 vectorize 对象
        self._problem_function_vectorized = None

    @abc.abstractmethod
    def problem_function(self, arg):
        pass

    def use(self, arg):
        # ✅ 惰性构建：仅在第一次 use() 调用时初始化
        if self._problem_function_vectorized is None:
            self._problem_function_vectorized = np.vectorize(
                self.problem_function,
                otypes=[np.float64]
            )
        re
turn self._problem_function_vectorized(arg)

class ProblemClass(ProblemClassBase):
    def problem_function(self, arg):
        if arg > 2:
            return arg + 1
        else:
            return arg - 1

class NestingClass:
    def __init__(self, problem_object):
        self.po = problem_object

    def make_problem(self, arg):
        return self.po.use(arg)

class MainClass:
    def __init__(self):
        self.problem_obj = ProblemClass()
        self.nesting_obj = NestingClass(self.problem_obj)

    def run(self, arg):
        return self.nesting_obj.make_problem(arg)

    @classmethod
    def run_multiproc(cls, arg):
        obj = cls()
        result = obj.run(arg)
        print(f"Process result for {arg}: {result}")
        return result  # 可通过 Queue / Pipe 返回

def run_parallel():
    proc = Process(target=MainClass.run_multiproc, args=(5,))
    proc.start()
    proc.join()

if __name__ == "__main__":
    run_parallel()

? 关键改进点说明：

立即学习“Python免费学习笔记（深入）”；

• self._problem_function_vectorized 初始化为 None，不触发 np.vectorize；
• use() 方法中检查缓存，未初始化则当场构建并保存，后续调用直接复用；
• 所有 np.vectorize 调用均发生在子进程内部（run_multiproc → cls() → use()），彻底规避跨进程传递问题。

⚠️ 注意事项：

• 若需支持高并发多调用（如每次 use() 输入不同 dtype），建议显式指定 otypes 并确保 problem_function 返回类型稳定；
• np.vectorize 本身不提升性能（仅为语法糖），真实加速应依赖 numba.jit、numpy.ufunc 原生操作或 dask 等；此处仅解决可序列化问题；
• 使用 pathos.multiprocess（基于 dill）虽能序列化更多对象，但 np.vectorize 实例仍属高危项，惰性初始化仍是更健壮、可移植的设计。

? 总结：多进程场景下，切勿将 np.vectorize、lambda、嵌套函数、闭包或任何非模块级可导入对象作为类/实例属性提前初始化。坚持“进程内构建、按需缓存”原则，即可兼顾简洁性与并行鲁棒性。