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]
複製代碼

參考

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。