针对同一数据集多进程执行独立函数的最优方案

你的场景简直是为多进程量身定做的——函数完全独立、无返回值、执行时长各异还允许部分失败，完全可以把它们分配到不同CPU核心并行执行，根本不用等前一个函数完成再启动下一个。下面给你两种实用的实现方案：

方式一：直接创建独立进程（最灵活） #

这种方式适合函数逻辑差异较大的场景，你可以手动给每个「函数-数据集」组合创建进程，启动后它们会立刻在后台并行运行，互不干扰。

示例代码 #

import multiprocessing
import time
import random

# 模拟你的数据处理函数：无返回值，结果保存到硬盘
def process_data_func1(dataset):
    try:
        print(f"启动process_data_func1，处理数据集大小：{len(dataset)}")
        # 模拟不同执行时长
        time.sleep(random.randint(2, 5))
        # 替换为实际的硬盘保存逻辑，比如：
        # with open("result_func1.txt", "w") as f:
        #     f.write(处理后的内容)
        print("process_data_func1执行完成，结果已保存")
    except Exception as e:
        print(f"process_data_func1执行失败：{str(e)}")
        # 失败可记录日志，不影响其他进程

def process_data_func2(dataset):
    try:
        print(f"启动process_data_func2，处理数据集大小：{len(dataset)}")
        time.sleep(random.randint(1, 3))
        print("process_data_func2执行完成，结果已保存")
    except Exception as e:
        print(f"process_data_func2执行失败：{str(e)}")

# 模拟一个可能失败的函数
def process_data_func3(dataset):
    try:
        print(f"启动process_data_func3，处理数据集大小：{len(dataset)}")
        time.sleep(random.randint(3, 6))
        # 故意抛出异常模拟失败
        raise ValueError("模拟函数执行出错")
        print("process_data_func3执行完成，结果已保存")
    except Exception as e:
        print(f"process_data_func3执行失败：{str(e)}")

if __name__ == "__main__":
    # 模拟你的大型数据集
    large_dataset = list(range(1000000))

    # 定义要执行的任务列表：(函数, 数据集)
    tasks = [
        (process_data_func1, large_dataset),
        (process_data_func2, large_dataset),
        (process_data_func3, large_dataset)
    ]

    # 创建并启动所有进程
    processes = []
    for func, data in tasks:
        p = multiprocessing.Process(target=func, args=(data,))
        processes.append(p)
        # 启动进程：调用后立刻返回，无需等待该进程完成
        p.start()

    # 等待所有进程执行完毕（如果主进程不需要立刻退出，可省略，但建议加上确保任务完成）
    for p in processes:
        p.join()

    print("所有任务处理结束（含失败任务）")

核心说明 #

• 每个p.start()会立刻启动对应进程，主进程继续创建下一个任务，实现真正的并行
• 函数内部的try-except确保单个函数失败不会崩溃整个程序，也不影响其他进程
• 完全满足「无需等待前一个完成即可启动下一个」的要求

方式二：使用进程池（更简洁） #

如果你的函数结构相对统一，或者想限制同时运行的进程数（避免占满所有CPU核心），可以用multiprocessing.Pool的apply_async方法，它也是非阻塞的，提交任务后立刻返回。

示例代码 #

import multiprocessing
import time
import random

# 复用上面的process_data_func1/func2/func3，此处省略重复代码...

if __name__ == "__main__":
    large_dataset = list(range(1000000))

    # 创建进程池，指定同时运行的进程数（比如4个，可根据CPU核心数调整）
    with multiprocessing.Pool(processes=4) as pool:
        # 异步提交所有任务
        task_results = []
        task_results.append(pool.apply_async(process_data_func1, args=(large_dataset,)))
        task_results.append(pool.apply_async(process_data_func2, args=(large_dataset,)))
        task_results.append(pool.apply_async(process_data_func3, args=(large_dataset,)))

        # 等待所有任务完成（主进程可在此期间做其他操作）
        for res in task_results:
            try:
                # get()会等待任务结束，捕获异常处理失败情况
                res.get()
            except Exception as e:
                print(f"任务执行失败：{str(e)}")

    print("所有任务处理完毕")

关键注意事项 #

1. 大数据集内存优化：多进程会复制数据集到每个进程的内存空间，如果数据集特别大，建议让每个进程直接从硬盘读取数据，或使用multiprocessing.Array/Manager实现共享内存，减少内存占用
2. Windows系统特殊要求：必须把主程序代码放在if __name__ == "__main__":块内，否则会重复创建进程导致错误
3. 失败处理：无论用哪种方式，都要在函数内部或主进程捕获异常，避免单个任务失败影响全局

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