问题背景
使用滑动窗口解决接口限流问题时需要统计连续时间内的请求数，可以用 Redis 对请求的信息设置过期时间，进而统计滑动窗口内的请求数，对阈值范围内的请求提供服务。但目前服务规模较小，且不需要 Redis 的大部分功能，不必要引入对 Redis 资源的依赖，因此基于 Python 字典实现 Redis TTL 功能
功能实现
学习 Redis 的实现，将 EXPIRE 提供的 key 的生命时间转化为 key 应过期时的 Unix Timestamp。在查询时检查 key 的过期时间，必要时进行清除操作
功能函数
1、 expire(key, ttl, now=None) 对指定的键设置生命时间，now是相对的起始时间 Unix Timestamp，默认是当前
2、 ttl(key, now=None) 返回剩余的生命时间
3、 setex(key, value, ttl) 赋值同时设置生命时间
4、 支持 dict 其他操作，例如 len(dict), dict[key], dict[key] = value, iter(dict), del dict[key]
用法
$ pip install pyttl
from pyttl import TTLDict
ttl_dict = TTLDict()
ttl_dict.setex('a', -1, 1)
ttl_dict.ttl('a')
ttl_dict['b'] = 'b_val'
实现方向
主要考虑了下面几个实现方法：
1、 学习 collections.DefaultDict 等，继承 dict 并重写部分类方法，出于性能优化的原因，dict 中部分方法并不是直接调用魔法方法实现，例如 update(), setdefault() 并不是直接调用 __setitem__进行存储，所以如果使用 dict 还需要重写 update() 等函数。DefaultDict 直接继承 dict 是因为它只增加了 __missing__的处理，并不影响核心功能
2、 继承抽象类 MutableMapping，实现必要的魔法方法（__init__, __len__, __iter__, __setitem__, __delitem__, __getitem__），通过 duck typing 实现字典功能
3、 继承封装了 MutableMapping 的 UserDict 类，数据通过 UserDict 的 data 对象存储
实现过程
实际采用了第3中方法： 继承 UserDict，并在 UserDict 的 data 对象中存储 (过期时间 timestamp, value) 的元组
1、 expire 实现:：将 data 中的值设置为 (过期时间 timestamp, value)，过期时间通过当前时间+生命时间计算得出
def expire(self, key, ttl, now):
self.data[key] = (now + ttl, value)
2、 ttl 实现：返回指定 key 剩余的生命时间，已过期的返回 -2 并将其删除，没有设置过期时间的返回 -1
def ttl(self, key, now=None):
expire, _value = self.data[key]
if expire is None:
# Persistent keys
return -1
elif expire <= now:
# Expired keys
del self[key]
return -2
return expire - now
3、 setex 实现：
def setex(self, key, ttl, value):
expire = time.time() + ttl
self.data[key] = (expire, value)
4、 iter 实现：遍历所有数据，使用生成器返回没有过期的值
def __iter__(self):
for k in self.data.keys():
ttl = self.ttl(k)
if ttl != -2:
yield k
5、 对象存取：存储 key, value 时默认将过期时间设置为 None，读取时先检查过期时间，如果没有过期会正常返回 value，否则会触发 KeyError
def __setitem__(self, key, value):
self.data[key] = (None, value)
def __getitem__(self, key):
self.ttl(key)
return self.data[key][1]
完整代码
最后还加上了并发控制
from collections import UserDict
from threading import RLock, Lock
import time
class TTLDict(UserDict):
def __init__(self, *args, **kwargs):
self._rlock = RLock()
self._lock = Lock()
super().__init__(*args, **kwargs)
def __repr__(self):
return '<TTLDict@%#08x; %r;>' % (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, now=None):
if now is None:
now = time.time()
with self._rlock:
expire, _value = self.data[key]
if expire is None:
# Persistent keys
return -1
elif expire <= now:
# Expired keys
del self[key]
return -2
return expire - now
def setex(self, key, ttl, value):
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]