如何在python的子类中并行化进程

问题描述

我正在处理一个项目，其中有一个 main 函数、一个名为 simulator 的类和一个名为 vehicle 的类。在主函数中，我调用：simulator.run()，它为 vehicle.run() 中的所有车辆调用 simulator.vehicle_list[]。

车辆必须计算轨迹，这是一个耗时的过程。这些进程彼此独立，我希望 vehicle.run() 计算在多个 CPU 内核上运行。

作为初学者，我创建了一个简单的项目（与上面的类名不同，但我希望足够直观）但我无法让它工作。 我的代码没有在子类中调用 self.calculate()。当我从主类中删除 if __name__ == '__main__': 语句时，出现错误：

'multi_subclass' object has no attribute '_closed'

我拥有的三个 python 文件：

multi_main.py

from multi_class import multi_class

main_class = multi_class()
main_class.calculate()

#main_class.calculate2()

multi_class.py

from multi_subclass import multi_subclass

class multi_class():
    def __init__(self):
        print("I am multi_class")
        subclass_list = []
        for i in range(10):
            subclass_list += [multi_subclass()]
        self.subclass_list = subclass_list
        
    def calculate(self):
        if __name__ == '__main__':
            processes = []
            for i in range(10):
                processes.append(self.subclass_list[i])
                self.subclass_list[i].start()
                
            [proc.join() for proc in processes]

multi_subclass.py

from multiprocessing import Process

class multi_subclass(Process):
    def __init__(self):
        print("I am multi_subclass")
        
    def calculate(self):
        print("Running massive calculations")
        self.member_variable = "Finished with calculation"
        
    def calculate2(self):
        print("I also want to be paralellized but NOT together with calculate!")
        
    def run(self):
        self.calculate()

解决方法

好的，所以这条语句 if __name__ == '__main__': 已经到位，这样如果主进程产生新进程，这些新进程也不会产生新进程。

所以，你的入口点应该有。

multi_main.py

from multi_class import multi_class
from timeit import default_timer


start = default_timer()

if __name__ == '__main__':

    main_class = multi_class()
    main_class.calculate()
    print(f"Your code ran in {default_timer() - start} seconds instead of 10 seconds because of multiprocessing.LOL")

multi_class.py calculate() 函数中有几个问题。

from multi_subclass import multi_subclass

class multi_class():
    def __init__(self):
        print("I am multi_class")
        subclass_list = []
        for i in range(10):
            subclass_list += [multi_subclass()]
        self.subclass_list = subclass_list

    def calculate(self):
        processes = []

        for i in range(10):
            # you were putting a different process in the list and never starting it but you were waiting for it to complete using `.join()` and starting a different process but never waiting for it to complete
            # fixed code below
            # processes.append(self.subclass_list[i])
            # self.subclass_list[i].start() 
            

            # create a process,put it in the list,start it
            process = self.subclass_list[i]
            processes.append(process)
            process.start()

        # wait for it to complete here
        [proc.join() for proc in processes]

multi_subclass.py

但问题是，当您使用 Process() 模块创建 multiprocessing 时，您提供了一个目标以在进程开始时运行，如下所示 -> Process(target=abc) abc 是一个函数。

但是在您的代码中，__init__ 不会调用 Process 类的 __init__，并且返回的对象是一个没有目标的简单对象。 我已经添加了代码。

from multiprocessing import Process
import time
class multi_subclass(Process):
    def __init__(self):
        print("I am multi_subclass")
        super().__init__(target=self.calculate)

    def calculate(self):
        print("Running massive calculations")
        time.sleep(1)
        self.member_variable = "Finished with calculation"


    def calculate2(self):
        print("I also want to be paralellized but NOT together with calculate!")

    def run(self):
        self.calculate()

输出

I am multi_class
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
I am multi_subclass
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Running massive calculations
Your code ran in 1.0859881550000001 seconds instead of 10 seconds because of multiprocessing.LOL