python - 在python中并行调用多个对象的方法

Question

我有两个类(class)。一个叫做algorithm，另一个叫做Chain。在 algorithm 中，我创建了多个链，它们将成为一系列采样值。我想在链级别并行运行采样。

也就是说，algorithm类实例化了n条链，我想运行属于Chain类的_sample方法，对于 algorithm 类中的每个并行链。

下面是一个示例代码，它尝试了我想做的事情。

我在这里看到了类似的问题:Apply a method to a list of objects in parallel using multi-processing ，但如函数 _sample_chains_parallel_worker 所示，此方法不适用于我的情况(我猜这是因为嵌套类结构)。

问题 1:为什么这不适用于这种情况？

_sample_chains_parallel 中的方法甚至没有并行运行。

问题 2:为什么？

问题 3:如何并行对这些链中的每一个进行采样？

import time
import multiprocessing

class Chain():

    def __init__(self):
        self.thetas = []

    def _sample(self):
        for i in range(3):
            time.sleep(1)
            self.thetas.append(i)

    def clear_thetas(self):
        self.thetas = []

class algorithm():

    def __init__(self, n=3):
        self.n = n
        self.chains = []

    def _init_chains(self):
        for _ in range(self.n):
            self.chains.append(Chain())

    def _sample_chains(self):
        for chain in self.chains:
            chain.clear_thetas()
            chain._sample()

    def _sample_chains_parallel(self):
        pool = multiprocessing.Pool(processes=self.n)
        for chain in self.chains:
            chain.clear_thetas()
            pool.apply_async(chain._sample())
        pool.close()
        pool.join()

    def _sample_chains_parallel_worker(self):

        def worker(obj):
            obj._sample()

        pool = multiprocessing.Pool(processes=self.n)
        pool.map(worker, self.chains)

        pool.close()
        pool.join()


if __name__=="__main__":
    import time

    alg = algorithm()
    alg._init_chains()

    start = time.time()
    alg._sample_chains()
    end = time.time()
    print "sequential", end - start

    start = time.time()
    alg._sample_chains_parallel()
    end = time.time()
    print "parallel", end - start

    start = time.time()
    alg._sample_chains_parallel_worker()
    end = time.time()
    print "parallel, map and worker", end - start

Answer 1

在 _sample_chains_parallel 中，您正在调用 chain._sample() 而不是仅仅传递函数:pool.apply_async(chain._sample())。因此，您将结果作为参数传递，而不是让 apply_async 计算它。

但删除 () 对您没有多大帮助，因为 Python 2 无法 pickle 实例方法(可能适用于 Python +3.5)。除非您在结果对象上调用 get()，否则它不会引发错误，所以如果您看到这种方法的时间很短，请不要高兴，那是因为它会立即退出并出现未引发的异常。

对于并行版本，您必须将 worker 重新定位到模块级别，并分别将其称为 pool.apply_async(worker (chain,)) pool。 map ( worker ，self.chains)。

请注意，您忘记了 _sample_chains_parallel_worker 的 clear_thetas()。无论如何，更好的解决方案是让 Chain._sample 负责调用 self._clear_thetas()。