1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
import os
import threading
class _HashedSeq(list):
__slots__ = 'hashvalue'
def __init__(self, tup, hash=hash):
self[:] = tup
self.hashvalue = hash(tup)
def __hash__(self):
return self.hashvalue
def _make_key(args, kw, typed=False):
key = args
if kw:
kwd_mark = (object(),)
key += kwd_mark
for item in kw.items():
key += item
if typed:
key += tuple(type(v) for v in args)
if kw:
key += tuple(type(v) for v in kw.values())
elif len(key) == 1 and type(key[0]) in {int, str}:
return key[0]
return _HashedSeq(key)
class ThreadSingletonMeta(type):
_instances = {}
_lock = threading.RLock()
def __call__(cls, *args, **kwargs):
key = (cls, threading.get_ident(), _make_key(args, kwargs))
with cls._lock:
if key not in cls._instances:
instance = super().__call__(*args, **kwargs)
cls._instances[key] = instance
return cls._instances[key]
class ProcessSingletonMeta(type):
_instances = {}
_lock = threading.RLock()
def __call__(cls, *args, **kwargs):
key = (cls, os.getpid(), _make_key(args, kwargs))
with cls._lock:
if key not in cls._instances:
instance = super().__call__(*args, **kwargs)
cls._instances[key] = instance
return cls._instances[key]
class SingletonMeta(type):
_instances = {}
_lock = threading.RLock()
def __call__(cls, *args, **kwargs):
key = (cls, _make_key(args, kwargs))
with cls._lock:
if key not in cls._instances:
instance = super().__call__(*args, **kwargs)
cls._instances[key] = instance
return cls._instances[key]
if __name__ == '__main__':
class A(metaclass=SingletonMeta):
def __init__(self):
print("A.__init__")
a1 = A()
a2 = A()
|