如何让multiprocessing进程池感知全局变量的变更？

在使用Python的multiprocessing进程池时遇到变量同步问题：

• 已知子进程修改的变量不会同步到主进程，但实际发现无论主进程在进程池启动前还是启动后修改全局变量，子进程都无法感知到变更。
• 简化示例代码：

import multiprocessing as mp

variable = 0

def double(i):
    return i * variable

def main():
    pool = mp.Pool()
    for result in pool.map(double, [1, 2, 3]):
        print(result)
    variable = 1

main()

即便把variable = 1移到pool = mp.Pool()之前，子进程始终只能读取到0。

• 实际场景需要子进程感知主循环中某个对象属性（如列表）的更新，曾尝试修改与进程关联的对象变量，虽有效但主进程性能骤降、CPU占用飙升，开销过大。

需求：高效低成本地让进程池中的子进程感知主进程修改的全局/对象变量，确保每次调用pool.map_async或pool.apply_async时，子进程使用更新后的变量。

1. 每次任务调用时显式传递更新后的变量 #

这是最直接且低开销的方式，适合变量更新频率不高或任务本身需要传递参数的场景。

示例代码： #

import multiprocessing as mp

def double(i, variable):
    return i * variable

def main():
    variable = 0
    pool = mp.Pool()
    # 第一次调用，传递初始值0
    results = pool.starmap(double, [(1, variable), (2, variable), (3, variable)])
    print(results)  # 输出 [0, 0, 0]
    
    variable = 1
    # 第二次调用，传递更新后的值1
    results = pool.starmap(double, [(1, variable), (2, variable), (3, variable)])
    print(results)  # 输出 [1, 2, 3]

if __name__ == "__main__":
    main()

2. 使用multiprocessing.Manager创建共享对象 #

如果需要子进程实时感知变量变化（而非每次任务传递），可以用Manager创建线程安全的共享对象，适合需要持续更新的场景。

示例代码（共享列表）： #

import multiprocessing as mp

def process_item(i, shared_list):
    # 读取共享列表的最新值
    return i * shared_list[0]

def main():
    with mp.Manager() as manager:
        shared_list = manager.list([0])  # 创建共享列表
        pool = mp.Pool()
        
        # 第一次任务
        results = pool.starmap(process_item, [(1, shared_list), (2, shared_list), (3, shared_list)])
        print(results)  # 输出 [0, 0, 0]
        
        # 主进程更新共享列表
        shared_list[0] = 1
        
        # 第二次任务，子进程读取到更新后的值
        results = pool.starmap(process_item, [(1, shared_list), (2, shared_list), (3, shared_list)])
        print(results)  # 输出 [1, 2, 3]

if __name__ == "__main__":
    main()

• 说明：Manager通过代理实现共享，开销远低于你之前尝试的关联对象变量方式，支持list、dict、Value等多种类型。

3. 使用multiprocessing.Value/Array（基础类型高效共享） #

如果共享的是单个数值或数组，Value和Array直接存储在共享内存中，比Manager代理的开销更低。

示例代码（共享整数）： #

import multiprocessing as mp

def double(i, shared_var):
    # 获取共享变量的当前值
    return i * shared_var.value

def main():
    # 创建共享整数，初始值0，'i'表示整数类型码
    shared_var = mp.Value('i', 0)
    pool = mp.Pool()
    
    results = pool.starmap(double, [(1, shared_var), (2, shared_var), (3, shared_var)])
    print(results)  # [0,0,0]
    
    # 主进程更新共享变量
    shared_var.value = 1
    
    results = pool.starmap(double, [(1, shared_var), (2, shared_var), (3, shared_var)])
    print(results)  # [1,2,3]
    
    pool.close()
    pool.join()

if __name__ == "__main__":
    main()

4. 重新初始化进程池（适合变量更新极少的场景） #

如果变量仅偶尔更新，可关闭旧进程池后重新创建——新进程会继承主进程当前的变量状态。

示例代码： #

import multiprocessing as mp

variable = 0

def double(i):
    return i * variable

def main():
    global variable
    # 第一次创建进程池，使用初始值0
    pool = mp.Pool()
    results = pool.map(double, [1,2,3])
    print(results)  # [0,0,0]
    pool.close()
    pool.join()
    
    # 更新全局变量
    variable = 1
    
    # 重新创建进程池，子进程继承新值
    pool = mp.Pool()
    results = pool.map(double, [1,2,3])
    print(results)  # [1,2,3]
    pool.close()
    pool.join()

if __name__ == "__main__":
    main()

• 缺点：创建销毁进程池有开销，仅适合变量更新频率极低的场景。

进程池的子进程在创建时会复制主进程的内存空间（写时复制机制），但子进程启动后，主进程的全局变量修改不会同步到子进程——因为两者拥有独立的地址空间。因此必须通过显式传递或共享内存/代理对象的方式实现变量同步。

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