Python3.x Python多进程：运行一个类的多个实例，将所有子进程保留在内存中

你要找的是sharing state between processes.根据文档，您可以创建shared memory，这对它可以存储的数据有限制，并且不是线程安全的，但提供了更好的速度和性能；或者可以使用server processes到managers.我们将使用后者，因为您希望共享用户定义数据类型的整个对象.请记住，使用管理器会影响代码的速度，这取决于向托管对象传递和接收的参数的复杂性.

经理、代理人和酸洗

如上所述，管理器创建服务器进程来存储对象，并允许通过代理访问对象.我回答了一个问题，更详细地介绍了它们的工作方式，以及如何创建合适的代理here.我们将使用链接答案中定义的相同代理，有some种变体.也就是说，我已经将__getattr__中的工厂功能替换为可以使用pickle进行酸洗的功能.这意味着您可以运行使用此代理创建的托管对象的实例方法，而无需使用multiprocess.结果是此修改的代理:

from multiprocessing.managers import NamespaceProxy, BaseManager
import types
import numpy as np


class A:
    def __init__(self, name, method):
        self.name = name
        self.method = method

    def get(self, *args, **kwargs):
        return self.method(self.name, args, kwargs)


class ObjProxy(NamespaceProxy):
    """Returns a proxy instance for any user defined data-type. The proxy instance will have the namespace and
    functions of the data-type (except private/protected callables/attributes). Furthermore, the proxy will be
    pickable and can its state can be shared among different processes. """

    def __getattr__(self, name):
        result = super().__getattr__(name)
        if isinstance(result, types.MethodType):
            return A(name, self._callmethod).get
        return result

解决方案

现在，我们只需要确保在创建monte_carlo的对象时，使用管理器和上述代理即可.为此，我们创建了一个名为create的类构造函数.应使用此函数创建monte_carlo的所有对象.这样，最终的代码如下所示:

from multiprocessing import Pool
from multiprocessing.managers import NamespaceProxy, BaseManager
import types
import numpy as np


class A:
    def __init__(self, name, method):
        self.name = name
        self.method = method

    def get(self, *args, **kwargs):
        return self.method(self.name, args, kwargs)


class ObjProxy(NamespaceProxy):
    """Returns a proxy instance for any user defined data-type. The proxy instance will have the namespace and
    functions of the data-type (except private/protected callables/attributes). Furthermore, the proxy will be
    pickable and can its state can be shared among different processes. """

    def __getattr__(self, name):
        result = super().__getattr__(name)
        if isinstance(result, types.MethodType):
            return A(name, self._callmethod).get
        return result


class monte_carlo:

    def __init__(self, ):
        self.x = np.ones((1000, 3))
        self.E = np.mean(self.x)
        self.Elist = []
        self.T = None

    def simulation(self, temperature):
        self.T = temperature
        for i in range(3000):
            self.MC_step()
            if i % 10 == 0:
                self.Elist.append(self.E)
        return

    def MC_step(self):
        x = self.x.copy()
        k = np.random.randint(1000)
        x[k] = (x[k] + np.random.uniform(-1, 1, 3))
        temp_E = np.mean(self.x)
        if np.random.random() < np.exp((self.E - temp_E) / self.T):
            self.E = temp_E
            self.x = x
        return

    @classmethod
    def create(cls, *args, **kwargs):
        # Register class
        class_str = cls.__name__
        BaseManager.register(class_str, cls, ObjProxy, exposed=tuple(dir(cls)))
        # Start a manager process
        manager = BaseManager()
        manager.start()

        # Create and return this proxy instance. Using this proxy allows sharing of state between processes.
        inst = eval("manager.{}(*args, **kwargs)".format(class_str))
        return inst


def proba(dE,pT):
    return np.exp(-dE/pT)


if __name__ == "__main__":
    Tlist = [1.1, 1.2, 1.3]
    N = len(Tlist)
    G = []

    # Create our managed instances
    for _ in range(N):
        G.append(monte_carlo.create())

    for _ in range(5):

        #  Run simulations in the manager server
        results = []
        with Pool(8) as pool:

            for i in range(N):  # this loop should be ran in multiprocess
                results.append(pool.apply_async(G[i].simulation, (Tlist[i], )))

            # Wait for the simulations to complete
            for result in results:
                result.get()

        for i in range(N // 2):
            dE = G[i].E - G[i + 1].E
            pT = G[i].T + G[i + 1].T
            p = proba(dE, pT)  # (proba is a function, giving a probability depending on dE)
            if np.random.random() < p:
                T_temp = Tlist[i]
                Tlist[i] = Tlist[i + 1]
                Tlist[i + 1] = T_temp

    print(Tlist)

这符合您想要的标准.它根本不创建任何副本，而是将simulation方法调用的所有参数在池中序列化，并发送到实际存储对象的管理器服务器.它在那里执行，结果(如果有)被序列化并在主进程中返回.所有这些，只使用内置的！

Output

[1.2, 1.1, 1.3]

Edit

由于您使用的是Linux，我建议您在if __name__ ...子句中使用multiprocessing.set_start_method将start方法设置为"spawn".这样做将确保子进程不能访问子句中定义的变量.