Python多进程字典追加问题：Jupyter运行冻结需重启内核

问题描述 #

在Jupyter Notebook中运行下述多进程Python代码时，程序出现冻结，必须重启内核才能恢复。代码意图是基于60×2001的DataFrame计算评分，将符合条件的结果存入共享字典dict_results中。

原代码：

import multiprocessing as mp

def search_regression_loop(data, row, dict_results):
    for row2 in range(2000):
        if row != row2:
            score = linear_model_pred_iloc(data.iloc[:,:51], row, row2)
            if (score > 0.30) and (score != 1):
                if (data.iloc[row,54] is not None) and (data.iloc[row2,54] is not None):
                    name =str(data.iloc[row,54]) + ' - ' + str(data.iloc[row2,54])
                    dict_results[name] = score                
                else:
                    name = str(data.iloc[row].name) + ' - ' + str(data.iloc[row2].name)
                    dict_results[name] = score

manager = mp.Manager()
dict_results = manager.dict()
procs = []

for row in tqdm(range(2000)):
    p = mp.Process(target=search_regression_loop, args=(tumor,row,dict_results))
    procs.append(p)
    p.start()

for p in procs:
    p.join()

问题原因 #

1. 进程数量过载：直接创建2000个进程，远超系统能稳定承载的进程数上限，导致CPU、内存资源耗尽，系统无响应。
2. DataFrame跨进程传递低效：每个进程都传递整个大DataFrame，引发大量数据拷贝，占用过多内存，还可能触发进程间数据同步异常。
3. 共享字典锁竞争：多个进程同时写入manager.dict()，会引发严重的锁竞争，导致进程阻塞无法推进。
4. Jupyter环境兼容性问题：Jupyter内核本身是单进程架构，大规模多进程操作容易出现资源管理混乱，加剧冻结概率。

解决方案 #

1. 使用进程池限制进程数量 #

用mp.Pool替代手动创建大量进程，进程数设置为CPU核心数（或略少），避免资源过载。

2. 优化数据传递逻辑 #

提前提取DataFrame的必要子集，减少跨进程传递的数据量；同时让每个进程先在本地存储结果，最后统一合并，避免共享字典的锁竞争。

3. 遵循多进程标准写法 #

添加if __name__ == "__main__": guard，这是Python多进程代码的规范要求，能避免Jupyter环境中的意外问题。

修改后的代码示例 #

import multiprocessing as mp
from tqdm import tqdm

def search_regression_loop(args):
    data, row = args
    local_results = {}
    # 提前提取需要的特征列，避免重复切片
    feature_data = data.iloc[:, :51]
    for row2 in range(2000):
        if row != row2:
            score = linear_model_pred_iloc(feature_data, row, row2)
            if (score > 0.30) and (score != 1):
                # 简化字符串拼接写法
                if data.iloc[row,54] is not None and data.iloc[row2,54] is not None:
                    name = f"{data.iloc[row,54]} - {data.iloc[row2,54]}"
                else:
                    name = f"{data.iloc[row].name} - {data.iloc[row2].name}"
                local_results[name] = score
    return local_results

if __name__ == "__main__":
    # 仅传递需要的列，减少数据拷贝
    data_subset = tumor.iloc[:, :55]
    # 使用CPU核心数作为进程数，平衡性能与资源占用
    num_processes = mp.cpu_count()
    with mp.Pool(num_processes) as pool:
        # 构造任务参数列表
        tasks = [(data_subset, row) for row in range(2000)]
        # 并行执行并展示进度
        results_list = list(tqdm(pool.imap(search_regression_loop, tasks), total=2000))
    
    # 合并所有进程的本地结果
    dict_results = {}
    for res in results_list:
        dict_results.update(res)

额外建议 #

如果Jupyter环境仍出现不稳定，建议将核心逻辑保存为独立.py脚本，通过终端运行，避免Jupyter内核的资源管理限制。

内容的提问来源于stack exchange，提问作者I'mStuckOnLine911