多进程Pool中向共享字典添加嵌套列表并保存JSON遇异常求助

我用Python的multiprocessing Pool运行多进程任务，注释掉向data_dict添加嵌套对象的代码时，所有逻辑运行正常；但启用这些代码后会出现异常。相关代码如下：

主进程代码 #

from multiprocessing.pool import Pool
from multiprocessing import Manager
from functools import partial
from tqdm import tqdm
import os
import json

manager = Manager()
lock = manager.Lock()
data_dict = manager.dict({data_subset: {}})
with Pool(processes=cpu_count()-2) as p:
        with tqdm(total=len(paths)) as pbar:
            for v in p.imap_unordered(
                partial(extract_video, root_dir=opt.data_path, dataset=dataset, output_path=opt.output_path, data_dict=data_dict, data_subset=data_subset),
                paths
            ):
                pbar.update()

print(data_dict.copy())
# Save the data_dict to a JSON file
with open(os.path.join(opt.output_path, "data_dict.json"), 'w') as json_file:
   print('writing to json')
   json.dump(data_dict.copy(), json_file)

子进程任务函数 #

import cv2

def extract_video(video, root_dir, dataset, output_path, data_dict, data_subset):
    try:
       # 生成图像裁剪的逻辑（省略）
       # ....
       for j, crop in enumerate(crops):
          image_path = os.path.join(output_path, id, "{}_{}.png".format(i, j))
          cv2.imwrite(image_path, crop)

          # Update data_dict
          label = 0  # 根据实际逻辑修改标签值

          # 尝试过不加锁的情况
          with lock:
             if id not in data_dict[data_subset]:                    
                 data_dict[data_subset][id] = {'label': label, 'list': []} # 也试过用manager.list()
             data_dict[data_subset][id]['list'] += [image_path]  # 也试过用append方法
             # 也试过直接赋值：data_dict[data_subset][id] = {'label': label, 'list': [image_path]}
        print(data_dict)
     except Exception as e:
        print("Error:", e)

问题原因 #

核心问题出在Manager代理对象的嵌套操作：

• Manager创建的data_dict是进程安全的代理字典，但它内部的嵌套字典（如data_dict[data_subset]）是普通Python字典，不具备进程安全特性，多进程同时修改会触发竞争条件。
• 即便加了锁，直接修改嵌套的普通字典/列表时，代理对象的同步机制可能因为无法感知内部容器的变化而抛出异常。

解决方案 #

1. 统一使用Manager代理容器 #

初始化时就确保所有嵌套的字典、列表都是Manager提供的进程安全代理对象：

manager = Manager()
lock = manager.Lock()
# 嵌套的字典也用manager.dict()创建
data_dict = manager.dict({data_subset: manager.dict()})

2. 修正子进程内的更新逻辑 #

在锁块中，新增的id对应的容器也要用Manager代理对象：

with lock:
    if id not in data_dict[data_subset]:                    
        # 用manager.dict和manager.list创建全代理结构
        data_dict[data_subset][id] = manager.dict({
            'label': label,
            'list': manager.list()
        })
    # 对代理列表使用append操作，避免+=可能引发的同步问题
    data_dict[data_subset][id]['list'].append(image_path)

3. 修复序列化与打印问题 #

Manager代理对象无法直接被json.dump序列化，打印时也可能出现异常，需要先转换为普通Python容器：

# 递归转换Manager代理容器为普通容器
def convert_to_plain(obj):
    if isinstance(obj, dict):
        return {k: convert_to_plain(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_to_plain(item) for item in obj]
    else:
        return obj

# 打印和序列化前转换
plain_data = convert_to_plain(data_dict)
print(plain_data)

with open(os.path.join(opt.output_path, "data_dict.json"), 'w') as json_file:
   print('writing to json')
   json.dump(plain_data, json_file)

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