Python 3.10多进程下如何安全向同一列表追加DataFrame

方法1：使用concurrent.futures.ProcessPoolExecutor（推荐） #

这种方式无需手动处理进程同步，让任务直接返回结果，由主进程统一收集，从根源避免共享列表的数据损坏问题。

改造思路 #

1. 将单日处理逻辑封装为子任务，或直接提交两个导出函数的独立任务
2. 利用ProcessPoolExecutor并行执行所有任务
3. 主进程单线程遍历执行结果，追加到列表（天然线程安全）

代码示例 #

import pandas as pd
from concurrent.futures import ProcessPoolExecutor

# 替换为你实际的函数实现
def ExportarMODml(n2):
    return pd.DataFrame({'module_ml': [n2, n2*2]})

def ExportarMODdl(n2):
    return pd.DataFrame({'module_dl': [n2+1, n2+2]})

def obtenern2(days):
    result_list = []
    # 根据CPU核心数调整max_workers，避免资源浪费
    with ProcessPoolExecutor(max_workers=4) as executor:
        tasks = []
        for d in days:
            # 替换为你生成n2的实际逻辑
            n2 = d * 3
            # 提交两个导出函数的并行任务
            tasks.append(executor.submit(ExportarMODml, n2))
            tasks.append(executor.submit(ExportarMODdl, n2))
        
        # 遍历完成的任务，收集结果
        for future in tasks:
            df = future.result()
            result_list.append(df)
    
    # 合并所有DataFrame
    final_df = pd.concat(result_list, ignore_index=True)
    return final_df

# 必须在__main__下启动多进程（Windows系统强制要求）
if __name__ == '__main__':
    test_days = [1,2,3,4,5]
    final_result = obtenern2(test_days)
    print(final_result)

方法2：使用multiprocessing.Queue手动管控结果 #

如果需要更底层的进程控制，可以用队列传递结果，主进程统一从队列中取出并追加到列表。

代码示例 #

import pandas as pd
import multiprocessing

def worker(d, queue):
    # 替换为你生成n2的实际逻辑
    n2 = d * 3
    df_ml = ExportarMODml(n2)
    df_dl = ExportarMODdl(n2)
    # 将结果放入队列
    queue.put(df_ml)
    queue.put(df_dl)

def obtenern2(days):
    result_list = []
    result_queue = multiprocessing.Queue()
    processes = []

    # 创建并启动进程
    for d in days:
        p = multiprocessing.Process(target=worker, args=(d, result_queue))
        processes.append(p)
        p.start()
    
    # 等待所有进程执行完毕
    for p in processes:
        p.join()
    
    # 从队列取出所有结果
    while not result_queue.empty():
        df = result_queue.get()
        result_list.append(df)
    
    final_df = pd.concat(result_list, ignore_index=True)
    return final_df

if __name__ == '__main__':
    test_days = [1,2,3,4,5]
    final_result = obtenern2(test_days)
    print(final_result)

关键注意事项 #

• 必须在if __name__ == '__main__':代码块内启动多进程，这是Python跨平台运行的强制要求
• 禁止在多进程中直接修改全局列表：每个进程拥有独立内存空间，修改不会同步到主进程
• 如果导出函数依赖外部资源（如文件、数据库连接），需在每个进程内独立初始化，避免资源冲突

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