使用多处理在类方法的 while 循环内运行函数答案

【问题标题】：Use multiprocessing to run functions inside a while loop in a class method使用多处理在类方法的 while 循环内运行函数
【发布时间】：2020-12-24 08:29:43
【问题描述】：

我有一种方法可以使用多种其他方法计算最终结果。它内部有一个while循环，不断检查新数据，如果收到新数据，它会运行其他方法并计算结果。这个 main 方法是唯一由用户调用的方法，它一直保持活动状态，直到程序关闭。基本结构如下：

class sample:
     def __init__(self):
           results = []
     def main_calculation(self):
           while True:
                 #code to get data
                 if newdata != olddata:
                       #insert code to prepare data for analysis
                       res1 = self.calc1(prepped_data)
                       res2 = self.calc2(prepped_data)
                       final = res1 + res2
                       self.results.append(final)

我想并行运行 calc1 和 calc2，这样可以更快地得到最终结果。但是，我不确定如何以这种方式实现多处理，因为我没有使用__main__ 保护。有没有办法并行运行这些进程？

这可能不是这段代码的最佳组织，但对于我正在运行的实际计算来说，这是最简单的，因为有必要从不同的文件导入和运行这段代码。但是，如果这不是可挽救的结构，我可以重构代码。

【问题讨论】：

您能否澄清一下您认为__main__ 守卫与多处理有什么关系？我不太确定我是否了解您真正想知道的内容，但我希望找到答案会让您更清楚。
@DavidZ 你好！到目前为止，在我发现的所有示例中，用于启动并行进程的行都在主守卫内，并且我发现了对不使用时出现的错误的引用：stackoverflow.com/questions/47705228/… 我得到了这个当我尝试从 while 循环内部并行启动进程时也会出错。谢谢！
好的，谢谢，这确实有助于澄清事情。您能否编辑您的问题以使其包含该信息？我会看看我能不能想出一个答案。

标签： python parallel-processing multiprocessing

【解决方案1】：

根据the documentation，您需要使用__main__ 守卫的原因是，当您的程序创建multiprocessing.Process 对象时，它会启动Python 解释器的全新副本，该副本将导入你程序的模块。如果导入你的模块调用 multiprocessing.Process() 本身，那将创建 Python 解释器的另一个副本，它解释你的代码的另一个副本，依此类推，直到你的系统崩溃（或者实际上，直到 Python 遇到不可重入的部分多处理代码）。

在程序的主模块中，通常调用顶层的一些代码，检查__name__ == '__main__' 是您可以判断程序是第一次运行还是作为子进程运行的方式。但是在不同的模块中，顶层可能没有任何代码（除了定义），在这种情况下不需要使用保护，因为可以安全地导入模块而无需启动新进程。

换句话说，这是危险的：

import multiprocessing as mp

def f():
    ...

p = mp.Process(target=f)
p.start()
p.join()

但这是安全的：

import multiprocessing as mp

def f():
    ...

def g():
    p = mp.Process(target=f)
    p.start()
    p.join()

这也是安全的：

import multiprocessing as mp

def f():
    ...

class H:
    def g(self):
        p = mp.Process(target=f)
        p.start()
        p.join()

因此，在您的示例中，您应该能够在函数中直接创建 Process 对象。

但是，我建议在该类的文档中明确说明该方法会创建一个Process，因为使用它的人（可能是您）需要知道在顶层调用该方法是不安全的的一个模块。就好像这样做，也属于“危险”的范畴：

import multiprocessing as mp

def f():
    ...

class H:
    def g(self):
        p = mp.Process(target=f)
        p.start()
        p.join()

H().g()  # this creates a Process at the top level

您还可以考虑另一种方法，让调用者完成所有流程创建。在这种方法中，您的sample 类构造函数或main_calculation() 方法都可以接受Pool 对象，它可以使用该池中的进程进行计算。例如：

class sample:
    def main_calculation(self, pool):
        while True:
            if newdata != olddata:
                res1_async = pool.apply_async(self.calc1, [prepped_data])
                res2_async = pool.apply_async(self.calc2, [prepped_data])
                res1 = res1_async.get()
                res2 = res2_async.get()
                # and so on

如果发生许多不同的计算，这种模式还可以让您的程序更有效地使用资源，因为它们都可以使用相同的进程池。

【讨论】：