使用concurrent.futures.ProcessPoolExecutor时全局字典无法同步问题

使用concurrent.futures.ProcessPoolExecutor时，尝试让工作进程修改全局作用域的var字典，工作进程内可以正常修改该字典，但所有进程完成后，主进程打印var却得到空字典，实际输出与预期不符。

原代码 #

from time import sleep
import concurrent.futures as cf

tasks = ["aaa", "bbb", "ccc"]

def my_worker(scan_task):
    global var
    print(scan_task)
    var[scan_task] = 0
    print(var)
    sleep(1)

var = {}

if __name__ == '__main__':
    with cf.ProcessPoolExecutor() as executor:
        futures = [executor.submit(my_worker, scan_task) for scan_task in tasks]
    cf.wait(futures)
    print("finished")
    print(var)

实际输出 #

aaa
{'aaa': 0}
bbb
{'aaa': 0, 'bbb': 0}
ccc
{'aaa': 0, 'bbb': 0, 'ccc': 0}
finished
{}

期望输出 #

aaa
{'aaa': 0}
bbb
{'aaa': 0, 'bbb': 0}
ccc
{'aaa': 0, 'bbb': 0, 'ccc': 0}
finished
{'aaa': 0, 'bbb': 0, 'ccc': 0}

问题原因 #

多进程之间的内存是完全隔离的：每个工作进程启动时会复制主进程的内存空间（包括var字典），工作进程内修改的只是自己副本中的字典，和主进程的var是完全独立的两个对象，因此主进程的var不会被工作进程的修改影响。

解决方案 #

方案1：工作进程返回结果，主进程汇总 #

修改工作函数，让它返回当前任务对应的键值对，主进程收集所有任务的结果后合并到字典中：

from time import sleep
import concurrent.futures as cf

tasks = ["aaa", "bbb", "ccc"]

def my_worker(scan_task):
    print(scan_task)
    sleep(1)
    return (scan_task, 0)  # 返回任务对应的键值对

if __name__ == '__main__':
    var = {}
    with cf.ProcessPoolExecutor() as executor:
        futures = [executor.submit(my_worker, scan_task) for scan_task in tasks]
        # 遍历完成的任务，合并结果到var
        for future in concurrent.futures.as_completed(futures):
            key, value = future.result()
            var[key] = value
    print("finished")
    print(var)

方案2：使用跨进程共享字典 #

如果需要在工作进程中直接修改共享字典，可以使用multiprocessing.Manager创建支持跨进程访问的字典对象：

from time import sleep
import concurrent.futures as cf
from multiprocessing import Manager

tasks = ["aaa", "bbb", "ccc"]

def my_worker(scan_task, shared_var):
    print(scan_task)
    shared_var[scan_task] = 0
    print(shared_var)
    sleep(1)

if __name__ == '__main__':
    with Manager() as manager:
        var = manager.dict()  # 创建跨进程共享的字典
        with cf.ProcessPoolExecutor() as executor:
            futures = [executor.submit(my_worker, scan_task, var) for scan_task in tasks]
            concurrent.futures.wait(futures)
        print("finished")
        print(dict(var))  # 转换为普通字典打印

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