使用多处理模块创建的进程之间的python类对象共享

Question

如何创建可由工作进程修改的类的 python 共享对象。我使用 multiprocessing.Process 模块创建了工作进程。我对 multiprocessing.Manager 模块有一些了解。谁能举例说明如何将我的类(class)注册到管理器、启动管理器并创建我的类(class)的共享对象。

Answer 1

这是一个例子:

from multiprocessing import Process, Pool
from multiprocessing.managers import BaseManager


class MySharedClass(object):
    stored_value = 0
    def get(self):
        return self.stored_value

    def set(self, new_value):
        self.stored_value = new_value
        return self.stored_value


class MyManager(BaseManager):
    pass


MyManager.register('MySharedClass', MySharedClass)

def worker ( proxy_object, i):
    proxy_object.set( proxy_object.get() + i )
    print ("id %d, sum %d" %(i, proxy_object.get()))
    return proxy_object


if __name__ == '__main__':
    manager = MyManager()
    manager.start()
    shared = manager.MySharedClass()

    pool = Pool(5)
    for i in range(33):
        pool.apply(func=worker, args=(shared, i))
    pool.close()
    pool.join()
    print "result: %d" % shared.get()

id 0，总和 0
id 1，sum 1
id 2，总和 3
...
编号 31，总数 496
编号 32，总和 528
结果:528

另一种变体(从未在实际项目中使用过):

from multiprocessing import Process, Pool
from multiprocessing.managers import BaseManager, NamespaceProxy


class MySharedClass(object):
    def __init__(self):
        self.stored_value = 0

    def get(self):
        return self.stored_value

    def set(self, new_value):
        self.stored_value = new_value
        return self.stored_value


class MyManager(BaseManager):
    pass

class MyProxy(NamespaceProxy):
    _exposed_ = ('__getattribute__', '__setattr__', '__delattr__')# add 'get' to use get


    #def get(self):
    #    callmethod = object.__getattribute__(self, '_callmethod')
    #    return callmethod('get')

MyManager.register('MySharedClass', MySharedClass, MyProxy)

def worker ( proxy_object, i):
    proxy_object.stored_value =  proxy_object.stored_value + i
    print ("id %d, sum %d" %(i, proxy_object.stored_value))
    return proxy_object


if __name__ == '__main__':
    manager = MyManager()
    manager.start()
    shared = manager.MySharedClass()
    print shared.stored_value

    pool = Pool(5)
    for i in range(33):
        pool.apply(func=worker, args=(shared, i))
    pool.close()
    pool.join()
    print "result: %d" % shared.stored_value