非主模块启动Python Multiprocessing：多模块场景实现可行性问询

Python的多进程机制并没有要求必须在主模块直接启动子进程，你完全可以把进程管理的逻辑封装到master模块中，只要处理好主模块的if __name__ == '__main__'防护就可以了——这是Windows系统下避免子进程重复执行主模块代码的关键，Linux/macOS下虽然影响较小，但加上这个判断会让代码更规范、跨平台性更好。

下面是完整的三个模块示例代码，完全符合你的需求：

Worker.py（定义工作单元） #

import time
import random

class Worker(object):
    @staticmethod
    def do_work(task_id):
        """模拟Worker执行的具体任务"""
        print(f"Worker 开始处理任务 {task_id}")
        # 随机休眠1-3秒，模拟耗时操作
        time.sleep(random.uniform(1, 3))
        print(f"Worker 完成任务 {task_id}")

master.py（封装进程启动逻辑） #

import multiprocessing
from Worker import Worker

class Master(object):
    def run(self, num_workers=3):
        """启动指定数量的Worker进程并等待完成"""
        worker_processes = []
        
        # 创建并启动多个Worker进程
        for task_id in range(num_workers):
            # 绑定Worker的静态方法作为进程执行目标，传入任务ID参数
            process = multiprocessing.Process(
                target=Worker.do_work,
                args=(task_id,)
            )
            worker_processes.append(process)
            process.start()
        
        # 等待所有Worker进程执行完毕
        for process in worker_processes:
            process.join()
        
        print("\n所有Worker任务已全部完成！")

MainTests.py（主运行脚本） #

from master import Master

if __name__ == '__main__':
    # 主脚本入口：实例化Master并启动任务
    task_master = Master()
    # 这里可以指定启动的Worker数量，比如4个
    task_master.run(num_workers=4)

关键注意点 #

• 主模块的if __name__ == '__main__'必须加：当multiprocessing启动子进程时，子进程会自动import主模块（也就是MainTests.py），如果没有这个判断，子进程会再次执行task_master.run()，导致无限创建进程（Windows下会直接抛出异常，Linux/macOS下也会出现不可预期的行为）。
• Worker方法的选择：示例中用了静态方法do_work，这是因为Windows系统下multiprocessing使用pickle序列化对象，实例方法可能会出现序列化失败的问题；如果一定要用实例方法，需要确保Worker实例可以被pickle，或者改用multiprocessing.Manager等机制，但静态方法的实现方式更简单可靠。
• 进程的等待：调用process.join()可以让主进程等待所有子进程完成后再继续执行，避免主进程提前退出导致子进程被强制终止。

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