Windows下Python多依赖并行函数代码结构优化方案咨询

你觉得所有逻辑必须塞进if __name__ == '__main__'块，是对Windows下多进程启动逻辑的误解。
Windows系统下Python多进程默认使用spawn启动模式，和Linux的fork模式不同，子进程启动时会重新导入整个主脚本文件。如果执行逻辑没有放在__main__保护块下，子进程导入脚本时会重复触发主流程，递归创建进程最终报错。但这个保护规则只需要包住「触发整个任务流运行的入口调用」，完全不需要把所有业务逻辑、函数定义都堆在这个块里。

把代码按「基础worker函数 -> 并行逻辑封装 -> 串行业务函数 -> 完整流程封装 -> 入口调用」分层定义，所有函数都放在顶层全局作用域，仅把最终的启动调用放在保护块内即可。
参考实现如下：

import concurrent.futures

# ----------------------
# 第一层：基础worker函数，全部定义在顶层，保证子进程可正常导入
# ----------------------
def worker_1(arg_A, arg_B):
    # 单条任务的处理逻辑
    ...
    return processed_item

def worker_2(input_from_worker1, arg_C):
    # 第二个并行阶段的单条任务逻辑
    ...
    return processed_item

# ----------------------
# 第二层：并行逻辑封装，也定义在顶层
# ----------------------
def parallelise_1(arg_A, arg_B):
    with concurrent.futures.ProcessPoolExecutor() as executor:
        # executor.map返回的迭代器可直接转list，无需手动循环append
        return list(executor.map(worker_1, arg_A, arg_B))

def parallelise_2(input_AA, arg_C):
    # 如果arg_C是固定参数，构造和输入长度一致的参数列表传给map即可
    arg_C_list = [arg_C] * len(input_AA)
    with concurrent.futures.ProcessPoolExecutor() as executor:
        return list(executor.map(worker_2, input_AA, arg_C_list))

# ----------------------
# 第三层：普通串行业务函数
# ----------------------
def some_other_function(input_BB, arg_D):
    # 串行处理逻辑
    ...
    return final_C

# ----------------------
# 第四层：完整业务流封装，同样放在顶层
# ----------------------
def run_full_pipeline(arg_A, arg_B, arg_C, arg_D):
    AA = parallelise_1(arg_A, arg_B)
    BB = parallelise_2(AA, arg_C)
    CC = some_other_function(BB, arg_D)
    return CC

# ----------------------
# 仅入口调用放在__main__保护块内
# ----------------------
if __name__ == '__main__':
    # 这里只做参数初始化、配置加载、入口调用即可，不需要堆业务逻辑
    # 示例参数
    arg_A = [...]
    arg_B = [...]
    arg_C = ...
    arg_D = ...
    
    result = run_full_pipeline(arg_A, arg_B, arg_C, arg_D)
    print(result)

如果后续并行函数变多，可以把重复的进程池初始化逻辑抽成公共装饰器，减少冗余代码：

from functools import wraps

def with_process_pool(worker):
    @wraps(worker)
    def wrapper(*iterable_args, max_workers=None):
        with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
            return list(executor.map(worker, *iterable_args))
    return wrapper

# 后续新增并行函数只需要加装饰器，不用重复写进程池初始化逻辑
@with_process_pool
def parallelise_1(arg_A, arg_B):
    return worker_1(arg_A, arg_B)

@with_process_pool
def parallelise_2(input_from_1, arg_C):
    return worker_2(input_from_1, arg_C)

• 所有worker函数、并行封装函数、业务流程函数必须定义在顶层全局作用域，不要嵌套定义在其他函数内部、也不要定义在__main__块内，否则子进程重新导入脚本时找不到对应函数，会抛出序列化错误。
• 传给worker的所有参数、worker返回的结果必须是可pickle序列化的，不要传打开的文件句柄、数据库连接、线程锁、局部嵌套函数这类无法跨进程传递的对象。
• 有前后依赖的并行步骤按顺序写即可，和普通串行代码的逻辑写法完全一致；如果是没有依赖关系的并行任务，可以放在同一个进程池内同时提交，减少进程创建销毁的开销，不用每步都新建进程池。

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