假裝優雅地實現定時快取裝飾器
參考資料
- Python 工匠：使用裝飾器的技巧
- 一日一技：實現有過期時間的LRU快取
這次的參考資料寫在前面，因為寫得真不錯！開始閱讀本篇分享前，建議先閱讀參考資料，如果還不能實現定時快取裝飾器，再繼續從這裡開始讀。
實現思路
功能拆分:
- 快取上次函式執行的結果一段時間。
- 把它封裝成裝飾器。
定時快取
眾所周知，python的functools庫中有lru_cache用於構建快取，而函式引數就是快取的key，因此，只要把快取空間設定為1，用時間值作為key，即可實現定時執行函式。細節就去看參考資料2吧，我這裡就不贅述了。
具體實現如下:
""" 定時執行delay_cache """ import time from functools import lru_cache def test_func(): print('running test_func') return time.time() @lru_cache(maxsize=1) def delay_cache(_): return test_func() if __name__ == "__main__": for _ in range(10): print(delay_cache(time.time()//1)) # 1s time.sleep(0.2)
程式輸出:
running test_func
1582128027.6396878
1582128027.6396878
running test_func
1582128028.0404685
1582128028.0404685
1582128028.0404685
1582128028.0404685
1582128028.0404685
running test_func
1582128029.0425367
1582128029.0425367
1582128029.0425367可以看到，test_func在距上次呼叫1s內直接輸出快取結果，呼叫間隔超過1s時test_func才會被真正執行。
手動實現快取需要用字典，這裡用lru_cache
裝飾器
裝飾器的作用呢，就是給函式戴頂帽子，然後函式該幹嘛幹嘛去，然而別人看見的已經不是原來的函式，而是戴帽子的函數了。哈哈。
@delay_cache(time.time()//1) # (midori)帽子
def test_func():
print('running test_func')
return time.time()一個錯誤的示範
實現這個delay_cache:
... import wrapt ... def delay_cache(t): @wrapt.decorator def wrapper(func, isinstance, args, kwargs): # 給func加快取 @lru_cache(maxsize=1) def lru_wrapper(t): return func() return lru_wrapper(t) return wrapper ...
執行這段程式，就會得到錯誤的結果……(嘿嘿)
test 1582129926.0
running test_func
1582129926.4459314
test 1582129926.0
running test_func
1582129926.6466658
test 1582129926.0
...可以看到，定時快取好像消失了一樣。原因是裝飾器返回的是wrapper函式，而引數t被wrapper函式排除在外了。用print列印t，就會發現t一直沒有變。
等等，如果t不變，那不應該是一直取快取結果嗎？
- 現實總是殘酷的，
wrapper函式返回的是lru_wrapper(t)，是一個結果，而不是lru_wrapper函式，於是可憐的lru_cache跟著執行完的lru_wrapper，被扔進了垃圾桶，從此被永遠遺忘。等到下一次執行到這裡，儘管新的t相同，但是lru_cache也是新的，它根本不記得自己曾經與t還有過一段美好的姻緣過往……
證據呢？如果你也和我一樣八卦的話，就去搞個全域性變數，在lru_wrapper首次執行的時候把它存下來，後面的呼叫就全靠這個全域性變數，然後輸出結果就不變了。(要記得只需要在lru_wrapper首次執行的時候把函式賦值給全域性變數！)
- 現實總是殘酷的×2，就算證明了
lru_cache和t隔世的姻緣，我們的需求也不會實現，因為之前說過，引數t被wrapper函式排除在外了。
如果不把t作為裝飾器的引數，而作為被裝飾函式的引數呢？功能倒是實現了，可是裝飾器失去了它的價值，而且每個使用者函式，比如這裡的test_func，都要加上時間計算，變成test_func(time.time()//1, ...):，到時候time模組滿天飛，難以直視，慘不忍睹。
正解
用類來做裝飾器，類例項化以後就可以一直相伴lru_cache左右，為它保駕護航。有關類裝飾器的內容看參考資料1
class DelayCache(object):
def __init__(self, delay_s):
self.delay_s = delay_s
@wrapt.decorator
def __call__(self, func, isinstance, args, kwargs):
self.func = func
self.args, self.kwargs = args, kwargs
hashable_arg = pickle.dumps((time.time()//self.delay_s, args, kwargs))
return self.delay_cache(hashable_arg)
@lru_cache(maxsize=1)
def delay_cache(self, _):
return self.func(*self.args, **self.kwargs)新的帽子做好了，給函式戴上試試看:
...
@DelayCache(1) # 快取 1s
def test_func(_):
print('running test_func')
return time.time()測試下效果:
if __name__ == "__main__":
for _ in range(10):
print(test_func(1)) # 只取定時快取
time.sleep(0.2)
# 測試結果:
# running test_func # 首次執行定時不是設定的1s，下面給出解決方案
# 1582132259.4029999
# 1582132259.4029999
# 1582132259.4029999
# running test_func
# 1582132260.0045283
# 1582132260.0045283
# 1582132260.0045283
# 1582132260.0045283
# 1582132260.0045283
# running test_func
# 1582132261.0072334
# 1582132261.0072334if __name__ == "__main__":
for i in range(10):
print(test_func(i)) # 每次都執行函式
time.sleep(0.2)
# 測試結果:
# running test_func
# 1582132434.0865102
# running test_func
# 1582132434.2869732
# running test_func
# 1582132434.4875488
# ...哈哈，這下終於搞定了。不過又冒出來2個問題:
首次執行的定時值並不是
1s。
函式每次開始計時的時間點都是隨機的，而快取更新卻依靠秒進位，所以首次執行的快取時間可能是0~1s內任意一個時間點到1s，所以不準。要解決這個問題，就要讓時間從0開始計時。我的做法是用一個self.start_time屬性記錄函式首次執行的時間，然後計算實際間隔的時候，用取到的時間減去這個記錄值，這樣起始時間就一定從0開始了。引數改變的時候計時沒有復位。
需要復位的地方就是執行delay_cache的地方，所以在delay_cache函式裡復位計時值即可。
另外，每次復位後，(time.time() - self.start_time)都重新從0開始累加，(time.time() - self.start_time) // self.delay_s的輸出會變成...0,1,0,0,0,0,1,0,0,0,0,1,0,0...，這樣就不能作為lru_cache的key來判定了，所以新增一個self.tick屬性，把狀態鎖住，變成...0,0,1,1,1,1,1,0,0,0,0,0,1,1...。
改動的地方直接看最終程式碼吧。
最終程式碼
import time
import pickle
import wrapt
from functools import lru_cache
class DelayCache(object):
def __init__(self, delay_s):
self.delay_s = delay_s
self.start_time = 0
self.tick = 0
@wrapt.decorator
def __call__(self, func, instance, args, kwargs):
self.func = func
self.args, self.kwargs = args, kwargs
if time.time() - self.start_time > self.delay_s:
self.tick ^= 1 # 狀態切換，相當於自鎖開關
hashable_arg = pickle.dumps((self.tick, args, kwargs))
return self.delay_cache(hashable_arg)
@lru_cache(maxsize=1)
def delay_cache(self, _):
self.start_time = time.time() # 計時復位
return self.func(*self.args, **self.kwargs)
@DelayCache(delay_s=1) # 快取1秒
def test_func(arg):
print('running test_func')
return arg
if __name__ == "__main__":
for i in [1, 1, 2, 3, 1, 1, 1, 1, 1, 1, 1, 1]:
print(test_func(i))
time.sleep(0.4)用@wrapt.decorator抵制套娃，用@lru_cache幹掉字典，程式碼變得異常清爽啊……
測試結果
running test_func
1
1
running test_func
2
running test_func
3
running test_func
1
1
1
running test_func
1
1
1
running test_func
1
1