Python multiprocessing多进程修改外部字典无效问题解决

multiprocessing的多进程基于独立操作系统进程实现，每个子进程拥有完全独立的内存空间：子进程启动时会复刻主进程的全局变量，但后续对变量的所有修改都只作用在自己的内存副本上，既不会同步回主进程，也不会同步给其他子进程。这就是主进程中results始终为空的核心原因。

你之前的两次调试没有暴露问题，是因为场景不涉及跨进程共享变量写入：

• 单进程调试时没有内存隔离，函数直接修改同一份全局字典，逻辑自然正常
• 打印hello的测试只执行无状态的输出操作，不需要跨进程共享数据，所以运行符合预期

另外原代码还有一处疏漏：启动进程后没有调用join()等待子进程执行完毕就直接打印结果，即便使用了共享变量，也可能因为进程未执行完成输出不完整的内容。

根据场景不同，推荐两种无冲突的多进程字典写入方式：

方案1：分片处理+主进程汇总（最推荐，无锁性能好） #

不需要让多个进程直接写同一个字典，把待处理的任务拆分成多个分片，每个进程独立处理自己的分片并返回结果，最后由主进程统一汇总到字典中。这种方式完全避免了跨进程写共享数据的竞态问题，性能远高于共享字典方案，是数据处理类任务的标准写法。

import multiprocessing
from functools import partial

# 每个进程处理分配到的key分片，返回键值对列表
def process_chunk(key_chunk, data_source):
    part_res = []
    for idx in key_chunk:
        part_res.append((idx, data_source[idx]))
    return part_res

if __name__ == "__main__": # Windows/macOS下多进程必须加这个入口判断，避免无限递归启动
    # 构造测试数据
    vals = [f"test seq {i}" for i in range(100)]
    data = {k:v for k,v in enumerate(vals)}
    results = {}
    core_num = multiprocessing.cpu_count()

    # 把字典key拆分为和核心数相等的分片
    all_keys = list(data.keys())
    chunk_size = len(all_keys) // core_num
    key_chunks = []
    for i in range(core_num):
        start = i * chunk_size
        # 最后一个分片处理剩余所有key，避免整除导致的遗漏
        end = start + chunk_size if i != core_num -1 else len(all_keys)
        key_chunks.append(all_keys[start:end])
    
    # 用进程池调度任务
    with multiprocessing.Pool(core_num) as pool:
        # 固定传入公共的data参数
        worker = partial(process_chunk, data_source=data)
        # 收集所有进程的处理结果
        chunk_outputs = pool.map(worker, key_chunks)
    
    # 主进程统一汇总所有结果
    for chunk in chunk_outputs:
        for k, v in chunk:
            results[k] = v
    
    print(results)

方案2：Manager托管共享字典（适合需要实时写入共享的场景） #

如果业务逻辑确实要求多个进程实时写入同一个字典，可以用multiprocessing.Manager创建托管字典——这个字典实际存储在独立的管理器进程中，所有子进程通过代理跨进程访问它，实现共享效果。注意写入时必须加锁，否则判断键是否存在、赋值两个操作之间可能被其他进程打断，导致数据错乱。

import multiprocessing

def my_func(shared_dict, write_lock, data_source):
    for i in range(len(data_source)):
        # 加锁保证判断+写入是原子操作，避免竞态
        with write_lock:
            if i not in shared_dict.keys():
                shared_dict[i] = data_source[i]

if __name__ == "__main__":
    vals = [f"test seq {i}" for i in range(100)]
    data = {k:v for k,v in enumerate(vals)}
    core_num = multiprocessing.cpu_count()

    # 初始化跨进程共享的字典和写锁
    manager = multiprocessing.Manager()
    results = manager.dict()
    lock = multiprocessing.Lock()

    processes = []
    for i in range(core_num):
        p = multiprocessing.Process(target=my_func, args=(results, lock, data))
        processes.append(p)
        p.start()
    
    # 等待所有进程执行完成
    for p in processes:
        p.join()
    
    print(dict(results)) # 转换为普通字典方便后续使用

注意：Manager托管字典的所有操作都需要走跨进程通信，性能比普通字典差很多，数据量较大时优先选择方案1。

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