为啥这个单例实现“不是线程安全的”？

Posted 2023-02-23

【中文标题】为啥这个单例实现“不是线程安全的”？

【英文标题】：Why is this singleton implementation "not thread safe"?为什么这个单例实现“不是线程安全的”？

【发布时间】：2018-11-07 01:56:11

【问题描述】：

1。 @Singleton 装饰器

我找到了一种优雅的方式来装饰 Python 类，使其成为singleton。该类只能产生一个对象。每个Instance() 调用都返回相同的对象：

class Singleton:
    """
    A non-thread-safe helper class to ease implementing singletons.
    This should be used as a decorator -- not a metaclass -- to the
    class that should be a singleton.

    The decorated class can define one `__init__` function that
    takes only the `self` argument. Also, the decorated class cannot be
    inherited from. Other than that, there are no restrictions that apply
    to the decorated class.

    To get the singleton instance, use the `Instance` method. Trying
    to use `__call__` will result in a `TypeError` being raised.

    """

    def __init__(self, decorated):
        self._decorated = decorated

    def Instance(self):
        """
        Returns the singleton instance. Upon its first call, it creates a
        new instance of the decorated class and calls its `__init__` method.
        On all subsequent calls, the already created instance is returned.

        """
        try:
            return self._instance
        except AttributeError:
            self._instance = self._decorated()
            return self._instance

    def __call__(self):
        raise TypeError('Singletons must be accessed through `Instance()`.')

    def __instancecheck__(self, inst):
        return isinstance(inst, self._decorated)

我在这里找到了代码： Is there a simple, elegant way to define singletons?

顶部的评论说：

[这是]一个非线程安全的帮助类，用于简化实现单例。

很遗憾，我没有足够的多线程经验来亲自了解“线程不安全”。

 

2。问题

我在多线程 Python 应用程序中使用这个 @Singleton 装饰器。我担心潜在的稳定性问题。因此：

有没有办法让这段代码完全线程安全？

如果上一个问题没有解决方案（或者它的解决方案太麻烦），我应该采取哪些预防措施来保证安全？

@Aran-Fey 指出装饰器编码错误。任何改进当然都非常感谢。

---

_{特此提供我当前的系统设置： >  Python 3.6.3 >  Windows 10，64 位}

【问题讨论】：

感谢您提供原始问题的链接；让这个答案很容易被否决......但说真的，这是一个糟糕的装饰器。

它似乎不起作用。

嗨@Aran-Fey，感谢您指出这一点。请随时对装饰器进行改进。我将不胜感激:-)

嗨@OlivierMelançon，到底是什么不工作？它（它 = 装饰器）似乎可以在我的系统上运行（但也许我在这里遗漏了一些东西）。但正如 Aran-Fey 刚刚指出的那样，也许应该改进装饰器 :-)

@K.Mulier 它确实有效......我只是觉得必须调用 Instance() 方法来获取单例很奇怪。我建议你看看this question，它展示了更整洁、更容易接受的单身人士的方式。由于它们被更广泛地使用，您会发现更容易获得有关它们的线程安全性的信息。

【参考方案1】：

我建议你选择一个更好的单例实现。 metaclass-based implementation 是最常用的。

至于线程安全，您的方法和上面链接中建议的任何方法都不是线程安全的：线程总是有可能读取到没有现有实例并开始创建实例，但是another thread does the same在存储第一个实例之前。

您可以使用with lock controller 来保护带有锁的基于元类的单例类的__call__ 方法。

import threading

lock = threading.Lock()

class Singleton(type):
    _instances = 

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            with lock:
                if cls not in cls._instances:
                    cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]


class SingletonClass(metaclass=Singleton):
    pass

作为suggested by se7entyse7en，您可以使用check-lock-check pattern。由于单例只创建一次，您唯一需要担心的是必须锁定初始实例的创建。虽然一旦完成，检索实例根本不需要锁定。出于这个原因，我们接受在第一次调用时重复检查，以便所有进一步的调用甚至不需要获取锁。

【讨论】：

太棒了！所以你将metaclass-based singleton implementation 与threading lock 结合起来，使整个事情成为线程安全的，对吧？

@K.Mulier 是的，你明白了

怎么样而不是装饰器@synchronized(lock) 只需使用with lock: ？

@OlivierMelançon with lock: 也在这里 :) 1 块简单地显示/分离受锁保护的内容，并且导入 functools 和包装没有开销。我们不使用@open_file()，而是使用with (open()):

@Troopers 这是check-lock-check pattern。它在我的答案中的代码下方进行了解释。

【参考方案2】：

我发布这个只是为了简化@OlivierMelançon 和@se7entyse7en 建议的解决方案：import functools 和包装没有开销。

import threading

lock = threading.Lock()

class SingletonOptmizedOptmized(type):
    _instances = 
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            with lock:
                if cls not in cls._instances:
                    cls._instances[cls] = super(SingletonOptmizedOptmized, cls).__call__(*args, **kwargs)
        return cls._instances[cls]

class SingletonClassOptmizedOptmized(metaclass=SingletonOptmizedOptmized):
    pass

区别：

>>> timeit('SingletonClass()', globals=globals(), number=1000000)
0.4635776
>>> timeit('SingletonClassOptmizedOptmized()', globals=globals(), number=1000000)
0.192263300000036

【参考方案3】：

如果您担心性能，您可以通过使用check-lock-check pattern 来最大程度地减少锁定获取来改进已接受答案的解决方案：

class SingletonOptmized(type):
    _instances = 

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._locked_call(*args, **kwargs)
        return cls._instances[cls]

    @synchronized(lock)
    def _locked_call(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(SingletonOptmized, cls).__call__(*args, **kwargs)

class SingletonClassOptmized(metaclass=SingletonOptmized):
    pass

这就是区别：

In [9]: %timeit SingletonClass()
488 ns ± 4.67 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

In [10]: %timeit SingletonClassOptmized()
204 ns ± 4 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

【讨论】：

这是一个我不知道的非常好的改进。我将它链接到我接受的答案中，让我知道你是否可以。

完全没有问题。越容易找到答案越好。

得到 NameError: name 'synchronized' is not defined。有什么要进口的吗？