python簡單模擬的本地輕量級緩存python
思路:緩存
# coding: utf-8 import weakref, collections class LocalCache(): notFound = object() # list dict等不支持弱引用,但其子類支持,故這裏包裝了下 class Dict(dict): def __del__(self): pass def __init__(self, maxlen=10): self.weak = weakref.WeakValueDictionary() self.strong = collections.deque(maxlen=maxlen) @staticmethod def nowTime(): return int(time.time()) def get(self, key): value = self.weak.get(key, self.notFound) if value is not self.notFound: expire = value[r'expire'] if self.nowTime() > expire: return self.notFound else: return value else: return self.notFound def set(self, key, value): # strongRef做爲強引用避免被回收 self.weak[key] = strongRef = LocalCache.Dict(value) # 放入定大隊列,彈出元素立刻被回收 self.strong.append(strongRef) # 裝飾器 from functools import wraps def funcCache(expire=0): caches = LocalCache() def _wrappend(func): @wraps(func) def __wrapped(*args, **kwargs): key = str(func) + str(args) + str(kwargs) result = caches.get(key) if result is LocalCache.notFound: result = func(*args, **kwargs) caches.set(key, {r'result': result, r'expire': expire + caches.nowTime()}) result = caches.get(key) return result return __wrapped return _wrappend # 測試函數 import time @funcCache(expire=300) def test_cache(v): # 模擬任務處理時常3秒 time.sleep(3) print('work 3s') return v print(test_cache(1)) print(test_cache(2)) print(test_cache(1)) print(test_cache(2))
輸出:app
work 3s
{'expire': 1524363279, 'result': 1}
work 3s
{'expire': 1524363282, 'result': 2}
{'expire': 1524363279, 'result': 1}
{'expire': 1524363282, 'result': 2}函數
可見第一次走具體處理函數,後面走的緩存測試