打包具有多处理代码的 python 模块

Question

我正在尝试使用 pyinstaller 将我的 python 项目打包成可执行文件。主模块包含用于多处理的代码。当我运行可执行文件时，只有在多处理部分之前的代码行会一次又一次地执行。它也不会抛出异常或退出程序。

主模块代码：

from Framework.ExcelUtility import ExcelUtility
from Framework.TestRunner import TestRunner
import concurrent.futures


class Initiator:

def __init__(self):
    self.exec_config_dict = {}
    self.test_list = []
    self.test_names = []
    self.current_test_set = []

def set_first_execution_order(self):
    # Code


def set_subsequent_execution_order(self):
    # Code

def kick_off_tests(self):
    '''Method to do Multi process execution'''
    if(__name__=="__main__"):
        with concurrent.futures.ProcessPoolExecutor(max_workers=int(self.exec_config_dict.get('Parallel'))) as executor:
            for test in self.current_test_set:
                executor.submit(TestRunner().runner,test)  ***This line is not being executed from the exe file.


initiator = Initiator()
initiator.get_run_info()
initiator.set_first_execution_order()
initiator.kick_off_tests()
while len(initiator.test_list) > 0:
    initiator.set_subsequent_execution_order()
    try:
        initiator.kick_off_tests()
    except BaseException as exception:
        print(exception)

Answer 1

根据问题定义，我假设您使用的是 ms-windows，并且主模块未命名为 __main__.py。

在这种情况下，multiprocessing 有一些特殊的准则：

确保新的 Python 解释器可以安全地导入主模块，而不会导致意外的副作用（例如启动新进程）。

和

应该使用if __name__ == '__main__'保护程序的“入口点”

所以，像这样更改主模块的最后一部分：

from multiprocessing import freeze_support

def kick_off_tests(self):
    '''Method to do Multi process execution'''
        with concurrent.futures.ProcessPoolExecutor(max_workers=int(self.exec_config_dict.get('Parallel'))) as executor:
            for test in self.current_test_set:
                executor.submit(TestRunner().runner,test)  


if __name__ == '__main__':
    freeze_support()
    initiator = Initiator()
    initiator.get_run_info()
    initiator.set_first_execution_order()
    initiator.kick_off_tests()
    while len(initiator.test_list) > 0:
        initiator.set_subsequent_execution_order()
        try:
            initiator.kick_off_tests()
        except BaseException as exception:
            print(exception)