safe_contrainer.py
812 Bytes
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
import threading
from typing import TypeVar, List, Dict, Generic
K = TypeVar('K') # 定义类型变量
V = TypeVar('V')
class SafeDict(Generic[K, V]):
def __init__(self):
self.lock = threading.Lock()
self.data: Dict[K, V] = {}
def put(self, key: K, value: V):
with self.lock:
self.data.update({key: value})
def remove(self, key: K):
with self.lock:
self.data.pop(key)
def values(self) -> List[V]:
with self.lock:
return list(self.data.values())
def get(self, key: K) -> V:
with self.lock:
return self.data.get(key)
if __name__ == '__main__':
test: SafeDict[int, str] = SafeDict()
test.put(1, '1')
print(test.get(1))
print(test.get(2))
t2 = {}
print(t2.get(1))