在所有子进程之间共享 Singleton 类的属性，而不将其作为函数参数传递

Question

我有一个 Singleton 类，我想在所有进程之间共享该 Singleton 类的属性，而不将其作为共享变量通过函数参数传递。

示例代码如下：

class Singleton():
   abc = {}
   def __call__():
      abc['key'] = ".com"

class myClass(metaclass=Singleton):
   def capslock(name):
      print name.upper()

if __name__==__main__:
   import multiprocessing as mp
   process1 = mp.Process(target=myClass.capslock, args=("stackoverflow"))

   process1.start()
   process1.join()

对于 print 语句，我需要 name.upper() + abc['key] 但所有子进程的 Singleton 属性将为空。

Answer 1

我很想对你的问题进行接近投票，因为尚不完全清楚你想要完成什么（请参阅我对你的问题发表的评论）。

但是，如果您尝试在创建单例时自动添加类属性

abc

，它将无法按照您建议的多处理方式工作，因为当使用

 将单例实例从主进程序列化/反序列化到子进程时pickle

，这会绕过正常的实例创建。 在类定义中定义的任何类属性都将按照最初定义进行腌制，但随后添加或修改的任何类属性都不会反映在子进程中。

以下代码演示了如何通过添加类属性来创建单例实例。但这也表明，当这样的实例被pickle到新的子进程时，更改后的类属性

x

将具有创建类时定义的值和类属性

abc

，该属性在类创建后动态添加，根本不存在：

class Singleton(type):
    _instances = {}

    def __call__(self, *args, **kwargs):
        if not self in self._instances:
            print('Creating singleton:')
            instance = super().__call__(*args, **kwargs)
            self._instances[self] = instance
            # Add class atribute
            instance.__class__.abc = {"key": "com"}
        return self._instances[self]

class myClass(metaclass=Singleton):
    x = 1 # class attribute

    @staticmethod
    def capslock(name):
        print(f'{name.upper()}, attribute x is {myClass.x}, attribute abc is {getattr(myClass, "abc", "missing")}')


if __name__== "__main__":
    import multiprocessing as mp

    # You need to crate an instance of myClass to get
    # class attribute abc. Here myClass.abc will be missing:
    print(f'attribute x is {myClass.x}, attribute abc is {getattr(myClass, "abc", "missing")}')

    myClass_singleton = myClass()
    # Now myClass.abc will exist:
    myClass.x = 2 # change dynamically
    print(f'attribute x is {myClass.x}, attribute abc is {getattr(myClass, "abc", "missing")}')

    # Verify we are getting singletons:
    myClass_singleton_another = myClass()
    print(myClass_singleton is myClass_singleton_another)

    # The class of the singleton and class attributes x and abc will be printed by
    # method capslock. When called by the main process, x will have the modified
    # value of 2 and abc will be a dictionary.
    myClass_singleton.capslock('main process')

    # When capslock is called by the child process, x will again be 1 and
    # abc will be missing entirely:
    process1 = mp.Process(target=myClass_singleton.capslock, args=("child process",))
    process1.start()
    process1.join()

打印：

attribute x is 1, attribute abc is missing
Creating singleton:
attribute x is 2, attribute abc is {'key': 'com'}
True
MAIN PROCESS, attribute x is 2, attribute abc is {'key': 'com'}
CHILD PROCESS, attribute x is 1, attribute abc is missing

解决方案

这里我们自定义

pickle

 序列化过程，以确保我们也序列化/反序列化类属性 

abc

。为此，我们定义了一个 mixin 类 

SingletonMixin

，我们的 

myClss

 类继承自它：

class Singleton(type):
    _instances = {}

    def __call__(self, *args, **kwargs):
        if not self in self._instances:
            print('Creating singleton:')
            instance = super().__call__(*args, **kwargs)
            self._instances[self] = instance
            # Add class atribute
            cls = self.__class__
            cls.abc = {"key": "com"}
        return self._instances[self]

class SingletonMixin:
    def __getstate__(self):
        return getattr(myClass, "abc", {}), self.__dict__

    def __setstate__(self, state):
        abc, __dict__ = state
        self.__dict__.update(__dict__)
        self.__class__.abc = abc

class myClass(SingletonMixin, metaclass=Singleton):

    def capslock(self, name):
        print(f'{name.upper()}, attribute abc is {getattr(myClass, "abc", "missing")}')

if __name__== "__main__":
    import multiprocessing as mp

    # You need to crate an instance of myClass to get
    # class attribute abc. Here myClass.abc will be missing:
    print(f'attribute abc is {getattr(myClass, "abc", "missing")}')

    myClass_singleton = myClass()
    # Now myClass.abc will exist:
    myClass.x = 2 # change dynamically
    print(f'attribute abc is {getattr(myClass, "abc", "missing")}')

    # Verify we are getting singletons:
    myClass_singleton_another = myClass()
    print(myClass_singleton is myClass_singleton_another)

    # The class of the singleton and class attributes x and abc will be printed by
    # method capslock. When called by the main process, x will have the modified
    # value of 2 and abc will be a dictionary.
    myClass_singleton.capslock('main process')

    # When capslock is called by the child process, x will again be 1 and
    # abc will be missing entirely:
    process1 = mp.Process(target=myClass_singleton.capslock, args=("child process",))
    process1.start()
    process1.join()

打印：

attribute abc is missing
Creating singleton:
attribute abc is {'key': 'com'}
True
MAIN PROCESS, attribute abc is {'key': 'com'}
CHILD PROCESS, attribute abc is {'key': 'com'}

Answer 2

您可以在

cached_classproperty

 的帮助下将 

__reduce__ 实现为类属性，以防止 pickle

 调用 

__new__

 和 

__setstate__

。

class myClass(metaclass=Singleton):
    @cached_classproperty
    def singleton_instance(cls) -> Self:
        return cls()

    @override
    def __reduce__(self) -> str | Tuple[Any, ...]:
        return (
            f"{type(self).__qualname__}.singleton_instance"
            if self is self.singleton_instance
            else super().__reduce__()
        )

请注意，如果您不手动调用

metaclass=Singleton

，则不需要 

myClass()

。只需将初始化移至

myClass.__init__

即可。

class myClass:
    @cached_classproperty
    def singleton_instance(cls) -> Self:
        return cls()

    @override
    def __reduce__(self) -> str | Tuple[Any, ...]:
        return (
            f"{type(self).__qualname__}.singleton_instance"
            if self is self.singleton_instance
            else super().__reduce__()
        )

   abc = {}
   def __init__(self):
      abc['key'] = ".com"

这比破解元类更Pythonic。