Python多进程全局变量设置：如何让子进程共享全局变量？

在Windows系统中，multiprocessing.Pool创建子进程时会重新导入主模块。你的set_globals函数仅在主进程的__main__代码块中执行，子进程导入模块时不会运行这部分代码，因此子进程内的data和my_class仍保持初始的None值，最终触发TypeError。

方案1：利用Pool的initializer初始化子进程全局变量 #

这是最直接的处理方式，让每个子进程启动时自动初始化所需的全局变量：

import numpy as np
from multiprocessing import Pool

data = None
my_class = None

class MyClass():
    def __call__(self, indx):
        return np.sum(data[:, indx])

def init_worker(n1, n2):
    global data, my_class
    data = np.arange(n1).reshape(n2, n2)
    my_class = MyClass()

def call_single_indx(indx):
    result = my_class(indx)
    return result

def launch_jobs(nmap=10, num_jobs=3, n1=400, n2=20):
    with Pool(processes=num_jobs, initializer=init_worker, initargs=(n1, n2)) as pool:
        result = pool.map(call_single_indx, range(nmap))
    result = np.array(result)
    return result

if __name__ == "__main__":
    result = launch_jobs()
    print(result)

方案2：绑定数据与可调用对象，脱离全局变量依赖 #

修改MyClass使其接收data作为初始化参数，避免依赖全局变量，直接将实例传递给子进程（注：大型numpy数组在进程间传递会触发拷贝，若要避免拷贝可结合共享内存方案）：

import numpy as np
from multiprocessing import Pool

class MyClass():
    def __init__(self, data):
        self.data = data
    def __call__(self, indx):
        return np.sum(self.data[:, indx])

def call_single_indx(args):
    indx, my_class = args
    return my_class(indx)

def launch_jobs(nmap=10, num_jobs=3):
    data = np.arange(400).reshape(20, 20)
    my_class = MyClass(data)
    with Pool(processes=num_jobs) as pool:
        result = pool.map(call_single_indx, [(i, my_class) for i in range(nmap)])
    result = np.array(result)
    return result

if __name__ == "__main__":
    result = launch_jobs()
    print(result)

方案3：共享内存存储超大型数组 #

如果数组体积极大，不想在进程间产生拷贝开销，可使用numpy共享内存方案：

import numpy as np
from multiprocessing import Pool, RawArray

class MyClass():
    def __init__(self, shape, dtype):
        self.shape = shape
        self.dtype = dtype
    def __call__(self, indx):
        data = np.frombuffer(shared_data, dtype=self.dtype).reshape(self.shape)
        return np.sum(data[:, indx])

def init_worker(arr):
    global shared_data
    shared_data = arr

def launch_jobs(nmap=10, num_jobs=3, n1=400, n2=20):
    dtype = np.int64
    data = np.arange(n1, dtype=dtype).reshape(n2, n2)
    # 创建共享内存数组
    shared_data = RawArray('q', data.size)  # 'q'对应int64类型
    # 将数据拷贝到共享内存
    np.frombuffer(shared_data, dtype=dtype)[:] = data.flatten()
    
    my_class = MyClass(data.shape, dtype)
    with Pool(processes=num_jobs, initializer=init_worker, initargs=(shared_data,)) as pool:
        result = pool.map(my_class, range(nmap))
    result = np.array(result)
    return result

if __name__ == "__main__":
    result = launch_jobs()
    print(result)

内容的提问来源于stack exchange，提问作者I.P. Freeley