Python多进程闭包变量共享问题：双文件场景差异排查及修复

场景1中skus是全局变量，进程池的子进程启动时会复制该全局变量的内存副本：

• 当使用chunksize=2时，同一个子进程会连续处理多个任务，复用同一个skus副本，因此后续任务能看到同进程内前面任务对skus的修改。
• 不同子进程的skus副本相互隔离，但同chunk内的任务共享副本，所以输出会出现同chunk内结果一致的情况。

场景2中skus是outer函数的局部变量，闭包call_func引用它：

• pathos.multiprocessing虽然支持序列化闭包，但每个子进程在接收闭包时，会绑定skus的独立初始副本，不同子进程的skus完全隔离。
• 处理2和3的子进程拿到的是skus的初始状态，无法看到处理0和1的子进程对skus的修改，因此输出不符合预期。

以下三种方案均可实现和场景1一致的输出：

方案1：将skus改为全局变量 #

和场景1逻辑对齐，让子进程复制全局变量的副本，同chunk内任务复用该副本：

from pathos.multiprocessing import Pool as ProcessPool
from decorate import timeit

# 将skus改为全局变量
skus = [i for i in range(4)]

@timeit
def outer():
    def call_func(i):
        skus[i] = i * 10
        return skus

    @timeit
    def process():
        with ProcessPool() as pool:
            res = pool.map(call_func, skus, chunksize=2)

        print("process result=", res)
    
    return process()

方案2：使用子进程初始化传递skus #

通过Pool的initializer和initargs，让每个子进程初始化时复制skus为局部全局变量，实现同chunk内任务复用：

from pathos.multiprocessing import Pool as ProcessPool
from decorate import timeit

@timeit
def outer():
    skus = [i for i in range(4)]
    # 子进程内的全局变量，用于复用副本
    child_skus = None

    def init_child(initial_skus):
        global child_skus
        # 子进程初始化时复制初始skus
        child_skus = initial_skus.copy()

    def call_func(i):
        child_skus[i] = i * 10
        return child_skus

    @timeit
    def process():
        # 传递初始化函数和参数
        with ProcessPool(initializer=init_child, initargs=(skus,)) as pool:
            res = pool.map(call_func, skus, chunksize=2)

        print("process result=", res)
    
    return process()

方案3：使用共享内存（实现真正进程间共享） #

如果需要所有子进程共享同一个skus对象（而非同进程内复用副本），可以使用multiprocessing.Manager创建共享列表：

from pathos.multiprocessing import Pool as ProcessPool
from multiprocessing import Manager
from decorate import timeit

@timeit
def outer():
    with Manager() as manager:
        # 创建可进程间共享的列表
        skus = manager.list([i for i in range(4)])

        def call_func(i):
            skus[i] = i * 10
            # 转换为普通列表返回（避免Manager对象序列化问题）
            return list(skus)

        @timeit
        def process():
            with ProcessPool() as pool:
                res = pool.map(call_func, skus, chunksize=2)

            print("process result=", res)
        
        return process()

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