Python dict 设置键值过时时间

时间 2019-11-17

原文原文链接

问题背景

使用滑动窗口解决接口限流问题时须要统计连续时间内的请求数，能够用 Redis 对请求的信息设置过时时间，进而统计滑动窗口内的请求数，对阈值范围内的请求提供服务。但目前服务规模较小，且不须要 Redis 的大部分功能，没必要要引入对 Redis 资源的依赖，所以基于 Python 字典实现 Redis TTL 功能python

功能实现

学习 Redis 的实现，将 EXPIRE 提供的 key 的生命时间转化为 key 应过时时的 Unix Timestamp。在查询时检查 key 的过时时间，必要时进行清除操做git

功能函数

expire(key, ttl, now=None) 对指定的键设置生命时间，now是相对的起始时间 Unix Timestamp，默认是当前
ttl(key, now=None) 返回剩余的生命时间
setex(key, value, ttl) 赋值同时设置生命时间
支持 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'
复制代码

实现方向

主要考虑了下面几个实现方法：github

学习 collections.DefaultDict 等，继承 dict 并重写部分类方法，出于性能优化的缘由，dict 中部分方法并非直接调用魔法方法实现，例如 update(), setdefault() 并非直接调用 __setitem__进行存储，因此若是使用 dict 还须要重写 update() 等函数。DefaultDict 直接继承 dict 是由于它只增长了 __missing__的处理，并不影响核心功能
继承抽象类 MutableMapping，实现必要的魔法方法（__init__, __len__, __iter__, __setitem__, __delitem__, __getitem__），经过 duck typing 实现字典功能
继承封装了 MutableMapping 的 UserDict 类，数据经过 UserDict 的 data 对象存储

实现过程

实际采用了第3中方法：继承 UserDict，并在 UserDict 的 data 对象中存储 (过时时间 timestamp, value) 的元组性能优化

expire 实现:：将 data 中的值设置为 (过时时间 timestamp, value)，过时时间经过当前时间+生命时间计算得出bash
```
def expire(self, key, ttl, now): 
	self.data[key] = (now + ttl, value)
复制代码
```

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
复制代码

setex 实现：app

def setex(self, key, ttl, value):
    expire = time.time() + ttl
    self.data[key] = (expire, value)
复制代码

iter 实现：遍历全部数据，使用生成器返回没有过时的值函数

def __iter__(self):
    for k in self.data.keys():
        ttl = self.ttl(k)
        if ttl != -2:
            yield k
复制代码

对象存取：存储 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]
复制代码