使用方式
t = TTL()
t.setex("k1", "v1", 50)
t.ttl("k1")
先看代码
class TTL(UserDict):
def __init__(self, *args, **kwargs):
self._rlock = RLock()
self._lock = Lock()
super().__init__(*args, **kwargs)
def __repr__(self):
return '<TTLDict:{} {} '.format(id(self), self.data)
def __expire__(self, key, ttl, now=None):
if now is None:
now = time.time()
with self._rlock:
_expire, value = self.data[key]
self.data[key] = (now + ttl, value)
def ttl(self, key: str, now=None):
if now is None:
now = time.time()
with self._rlock:
expire, value = self.data.get(key, (None, None))
if expire is None:
return -1
elif expire <= now:
del self[key]
return -2
return expire - now
def setex(self, key: str, value: str, ttl: int):
with self._rlock:
expire = time.time() + ttl
self.data[key] = (expire, value)
def __len__(self):
with self._rlock:
for key in list(self.data.keys()):
self.ttl(key)
return len(self.data)
def __iter__(self):
with self._rlock:
for k in self.data.keys():
ttl = self.ttl(k)
if ttl != -2:
yield k
def __setitem__(self, key, value):
with self._lock:
self.data[key] = (None, value)
def __delitem__(self, key):
with self._lock:
del self.data[key]
def __getitem__(self, key):
with self._rlock:
self.ttl(key)
return self.data[key][1]
参考:
https://juejin.im/post/5dc50059e51d4558a92eb27c
网友评论