Windows下Python multiprocessing动态类型序列化问题求解

问题背景 #

在Windows环境使用multiprocessing传递动态创建的类时，会触发Pickle序列化错误，相同代码在Linux/WSL环境可正常运行。当前需求是仅在主进程构建一次动态类（类构建逻辑耗时且为常量），子进程直接初始化该类，避免重复构建的资源消耗。

原代码及报错 #

原测试代码 #

import multiprocessing
class Handler:
    pass

handler = type('HandlerClass', (int, Handler),{"const": [123]})
def stream_process(H):
    h=H()
    print(h.const)

processes = [ multiprocessing.Process(target=stream_process, args=( handler,), name=f"server #{i}") for i in range(4) ]

for process in processes:
    process.start()

for process in processes:
    process.join()

报错信息 #

Traceback (most recent call last):
File "C:\Projects\winmp\test.py", line 18, in
process.start()
File "C:\Programs\Python\Python312\Lib\multiprocessing\process.py", line 121, in start
self._popen = self._Popen(self)
^^^^^^^^^^^^^^^^^
File "C:\Programs\Python\Python312\Lib\multiprocessing\context.py", line 224, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Programs\Python\Python312\Lib\multiprocessing\context.py", line 337, in _Popen
return Popen(process_obj)
^^^^^^^^^^^^^^^^^^
File "C:\Programs\Python\Python312\Lib\multiprocessing\popen_spawn_win32.py", line 94, in init
reduction.dump(process_obj, to_child)
File "C:\Programs\Python\Python312\Lib\multiprocessing\reduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <class 'main.HandlerClass'>: attribute lookup HandlerClass on main failed
Traceback (most recent call last):
File "", line 1, in
File "C:\Programs\Python\Python312\Lib\multiprocessing\spawn.py", line 108, in spawn_main
source_process = _winapi.OpenProcess(
^^^^^^^^^^^^^^^^^^^^
OSError: [WinError 87] The parameter is incorrect

绑定模块属性的失败尝试 #

尝试将动态类绑定到模块全局属性，但仍报错：

测试代码 #

import multiprocessing
import sys
class Handler:
    pass


def stream_process(H):
    h = H()
    print(h.from_bytes(b'a', "big"))
    print(h.const)


def main():
    global HandlerClass
    HandlerClass = type('HandlerClass', (int, Handler),{"const": [123]})
    setattr(sys.modules[__name__], "HandlerClass",HandlerClass )
    setattr(sys.modules['test'], "HandlerClass",HandlerClass )
    
    processes = [ multiprocessing.Process(target=stream_process, args=( HandlerClass,), name=f"server #{i}") for i in range(4) ]

    for process in processes:
        process.start()

    for process in processes:
        process.join()

报错信息 #

Traceback (most recent call last):
File "", line 1, in
File "C:\Programs\Python\Python312\Lib\multiprocessing\spawn.py", line 122, in spawn_main
exitcode = _main(fd, parent_sentinel)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Programs\Python\Python312\Lib\multiprocessing\spawn.py", line 132, in _main
self = reduction.pickle.load(from_parent)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: Can't get attribute 'HandlerClass' on <module 'test' from 'C:\Projects\winmp\test.py'>

问题根源 #

Windows下multiprocessing默认使用spawn模式：子进程会重新导入主模块，此时动态类尚未被创建，导致Pickle无法找到类的全局引用；而Linux/WSL使用fork模式，直接复制主进程内存空间，无需重新导入模块，因此可以正常访问动态类。绑定模块属性失败的原因是子进程启动时，main()函数尚未执行，模块中不存在对应的类属性。

解决方案 #

方案1：使用cloudpickle序列化动态类 #

cloudpickle扩展了标准Pickle的能力，可直接序列化动态创建的类，无需依赖全局引用。

1. 安装cloudpickle：

pip install cloudpickle

2. 修改后的代码：

import multiprocessing
import cloudpickle

class Handler:
    pass

def stream_process(H):
    h = H()
    print(h.const)

# 包装函数，用于反序列化传递的类
def wrapped_stream_process(serialized_class):
    target_class = cloudpickle.loads(serialized_class)
    stream_process(target_class)

if __name__ == '__main__':
    # 主进程仅创建一次动态类并序列化
    handler_class = type('HandlerClass', (int, Handler), {"const": [123]})
    serialized_handler = cloudpickle.dumps(handler_class)

    processes = [
        multiprocessing.Process(
            target=wrapped_stream_process,
            args=(serialized_handler,),
            name=f"server #{i}"
        ) for i in range(4)
    ]

    for process in processes:
        process.start()
    for process in processes:
        process.join()

方案2：自定义类的序列化逻辑 #

通过给动态类添加__reduce__方法，明确告诉Pickle如何序列化和还原类，同时确保子进程无需重复执行耗时的构建逻辑：

import multiprocessing

class Handler:
    pass

# 封装类的构建参数，避免子进程重复执行耗时逻辑
CLASS_ARGS = ('HandlerClass', (int, Handler), {"const": [123]})

def create_handler_class():
    return type(*CLASS_ARGS)

if __name__ == '__main__':
    # 主进程仅构建一次类
    handler_class = create_handler_class()
    # 给动态类添加__reduce__方法，指定还原逻辑
    handler_class.__reduce__ = lambda cls: (create_handler_class, ())

    def stream_process(H):
        h = H()
        print(h.const)

    processes = [
        multiprocessing.Process(
            target=stream_process,
            args=(handler_class,),
            name=f"server #{i}"
        ) for i in range(4)
    ]

    for process in processes:
        process.start()
    for process in processes:
        process.join()

此方案中，CLASS_ARGS存储类的构建参数，子进程通过create_handler_class快速重建类，无需重新执行原有的耗时构建逻辑。

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