使用N个worker通过multiprocessing处理任务列表的最佳方案及代码优化

现有代码存在的问题 #

• 参数传递错误：ProcessPoolExecutor.map()会将args中的每个元素直接作为单个参数传入do_stuff，你传入的每个元素是二元组，会导致do_stuff收到1个元组参数而非预期的2个参数，触发参数数量不匹配报错。
• 调用次数不符合要求：现有逻辑会对每个参数组调用一次do_stuff，如果你的data有N条数据，do_stuff就会被调用N次，完全不满足「最多执行5次、每次处理一批参数」的需求。
• 缺少多进程入口判断：Windows平台下运行Python多进程代码必须加if __name__ == "__main__"的入口限制，否则会无限递归启动子进程，直接报错。

优化后可直接运行的方案 #

实现思路 #

先把所有参数均匀拆分为最多5个批次，每个批次对应一个worker要处理的全部参数，每个worker仅调用一次do_stuff处理对应批次即可，刚好匹配调用次数要求。

完整代码 #

from concurrent.futures import ProcessPoolExecutor

def batch_process(func, batch_args, workers):
    with ProcessPoolExecutor(workers) as ex:
        res = ex.map(func, batch_args)
    return list(res)

def do_stuff(batch_params):
    '''单次执行，处理分配给自己的整批参数'''
    process_result = []
    for arg1, arg2 in batch_params:
        # 此处替换为你实际的单条参数处理逻辑
        process_result.append(f"处理完成：arg1={arg1}, arg2={arg2}")
    return process_result

def split_batches(all_items, max_batch):
    """将参数列表拆分为最多max_batch个非空批次"""
    batches = [[] for _ in range(max_batch)]
    for index, item in enumerate(all_items):
        batches[index % max_batch].append(item)
    # 过滤空批次，避免无效调用
    return [batch for batch in batches if batch]

def main():
    # 替换为你的实际数据源
    data = [1,2,3,4,5,6,7,8,9,10,11]
    # 生成所有单条参数组
    all_args = [(item, 123) for item in data]
    # 拆分为最多5个批次
    task_batches = split_batches(all_args, 5)
    # 执行多进程，do_stuff最多被调用5次
    results = batch_process(do_stuff, task_batches, 5)
    # 输出各批次处理结果
    for batch_id, batch_res in enumerate(results):
        print(f"批次{batch_id}处理结果：{batch_res}")

if __name__ == "__main__":
    main()

关键调整说明 #

• 新增批次拆分逻辑，保证do_stuff最多被调用5次，参数数量不足5个批次时自动过滤空批次，不会触发无效调用。
• 调整do_stuff入参为整批参数，单次启动后遍历处理批次内所有参数，符合业务要求。
• 新增多进程入口判断，兼容Windows、Linux、macOS全平台运行。
• 修复原参数传递错误问题，运行不会再报参数不匹配的异常。

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