|
@@ -0,0 +1,41 @@
|
|
|
+import functools
|
|
|
+import threading
|
|
|
+
|
|
|
+lock = threading.Lock()
|
|
|
+
|
|
|
+def synchronized(lock):
|
|
|
+ """ Synchronization decorator """
|
|
|
+ def wrapper(f):
|
|
|
+ @functools.wraps(f)
|
|
|
+ def inner_wrapper(*args, **kw):
|
|
|
+ with lock:
|
|
|
+ return f(*args, **kw)
|
|
|
+ return inner_wrapper
|
|
|
+ return wrapper
|
|
|
+
|
|
|
+class SingletonThreadsafe(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(SingletonThreadsafe, cls).__call__(*args, **kwargs)
|
|
|
+
|
|
|
+
|
|
|
+class testClass(metaclass=SingletonThreadsafe):
|
|
|
+ pass
|
|
|
+ def __init__(self, message: str = None):
|
|
|
+ self._message = message
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ x = testClass('hello')
|
|
|
+ y = testClass('yellow')
|
|
|
+
|
|
|
+ print(x==y)
|
|
|
+ print(x._message)
|
|
|
+ print(y._message)
|