假装优雅地实现定时缓存装饰器

Posted adjwang

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了假装优雅地实现定时缓存装饰器相关的知识,希望对你有一定的参考价值。

参考资料

  1. Python 工匠:使用装饰器的技巧
  2. 一日一技:实现有过期时间的LRU缓存

这次的参考资料写在前面,因为写得真不错!开始阅读本篇分享前,建议先阅读参考资料,如果还不能实现定时缓存装饰器,再继续从这里开始读。

实现思路

功能拆分:

  1. 缓存上次函数运行的结果一段时间。
  2. 把它封装成装饰器。

定时缓存

众所周知,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内直接输出缓存结果,调用间隔超过1stest_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函数,而参数twrapper函数排除在外了。print打印t,就会发现t一直没有变。
等等,如果t不变,那不应该是一直取缓存结果吗?

  • 现实总是残酷的,wrapper函数返回的是lru_wrapper(t),是一个结果,而不是lru_wrapper函数,于是可怜的lru_cache跟着执行完的lru_wrapper,被扔进了垃圾桶,从此被永远遗忘。等到下一次执行到这里,尽管新的t相同,但是lru_cache也是新的,它根本不记得自己曾经与t还有过一段美好的姻缘过往……
    证据呢?如果你也和我一样八卦的话,就去搞个全局变量,在lru_wrapper首次运行的时候把它存下来,后面的调用就全靠这个全局变量,然后输出结果就不变了。(要记得只需要在lru_wrapper首次运行的时候把函数赋值给全局变量!)
  • 现实总是残酷的×2,就算证明了lru_cachet隔世的姻缘,我们的需求也不会实现,因为之前说过,参数twrapper函数排除在外了。

如果不把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.0072334
if __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个问题:

  1. 首次运行的定时值并不是1s
    函数每次开始计时的时间点都是随机的,而缓存更新却依靠秒进位,所以首次运行的缓存时间可能是0~1s内任意一个时间点到1s,所以不准。要解决这个问题,就要让时间从0开始计时。我的做法是用一个self.start_time属性记录函数首次运行的时间,然后计算实际间隔的时候,用取到的时间减去这个记录值,这样起始时间就一定从0开始了。

  2. 参数改变的时候计时没有复位。
    需要复位的地方就是执行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_cachekey来判定了,所以添加一个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

以上是关于假装优雅地实现定时缓存装饰器的主要内容,如果未能解决你的问题,请参考以下文章

别再写满屏的爆爆爆炸类了,试试装饰器模式,这才是优雅的方式!!

FastAPI利用装饰器实现定时任务

FastAPI利用装饰器实现定时任务

FastAPI利用装饰器实现定时任务

Python之如何优雅的重试

别再写满屏的爆爆爆炸类了,试试装饰器模式,这才是优雅的方式。。