【Python】优雅的实现缓存功能
需求背景
需求背景简化一下大概如下图：
这个API要完成一些复杂的统计，需要查询很多次MySQL数据库，数据量又很大，造成这个接口的延迟很高，高峰的时候超过10秒以上。
我们暂时假定这个需求是合理的（事实上这个需求是可以从业务上优化的）。
第一层次的缓存
第一层优化很简单，直接在Http API接口层做缓存。
这个很容易想到，但是这个有比较大问题，因为这个接口本身延时比较高，如果在缓存失效的情况下，进来了很多的请求，系统可能就崩溃了。而这个发生的条件是比较容易满足的，因为时延比较高。
所以这是不可取的。
第二层次的缓存
外层接口的时延很高，但是底层公共查询函数每次的耗时其实就200毫秒左右，如果直接在这一层做缓存，那肯定可以满足我们的系统要求了。
实现也简单：
def get_data_dict(self, sql, params=[], cache: bool = False, timeout: int = 0):
"""查询，返回字典类型"""
def _get_data():
if params:
self.dict_cur.execute(sql, params)
else:
self.dict_cur.execute(sql)
return self.dict_cur.fetchall()
if not cache:
return _get_data()
# 开启了缓存
_redis = RedisPool()
cache_key = hash_id(sql + str(params))
cache_data = _redis.getKey(cache_key)
if cache_data: # 命中缓存
return json.loads(cache_data)
data = _get_data()
_redis.setKey(cache_key, json.dumps(data), timeout)
return data这样可以实现需求，但是这并不优雅，因为如果其他的函数需要函数，还要非常相似的一堆代码来实现！
所以我们应该实现成装饰器才是最佳的选择。
第三层次的缓存
写成一个装饰器并不难：
def CacheFunc(function):
"""对函数的返回结果进行缓存（使用redis）
缓存时可以指定redis连接对象、key前缀、有效期等
"""
@wraps(function)
def wrapper(*args, redis_connect=None, key_prefix='cf', expire_second=30, **kwargs):
"""
注意:
1. 如果函数执行时间过程，redis对象可能会超时
2. 注意kwargs中的参数不能和下面三个参数冲突
:param redis_connect redis.Redis() redis操作对象，如果该值为None则不进行缓存
:param key_prefix str 缓存key前缀
:param expire_second int 过期时间，单位：秒
"""
if redis_connect is None:
return function(*args, **kwargs)
key = "%s:%s:%s" % (key_prefix, function.__name__, md5(str(args) + str(kwargs)[8:24]))
data = redis_connect.get(key)
if data:
return json.loads(data)
data = function(*args, **kwargs)
redis_connect.set(key, json.dumps(data), ex=expire_second)
return data
return wrapper对应的单元测试代码：https://github.com/ibbd-dev/python-ibbd-algo/blob/c3ff9113fc6b3475fdde8be29547b578ca6aef24/ibbd_algo/decorator_test.py
这个实现应该说自己已经挺满意了，不过还是有几个问题：
- 强依赖于redis；
- 参数名可能会存在冲突；
- 上游函数如果返回None、0、空字符串、空列表等会导致击穿缓存。
特别是第三点缺陷是比较明显的，需要优化。
知道问题在哪里，优化并不难：
def CacheFunc(function):
"""对函数的返回结果进行缓存（使用redis）
缓存时可以指定redis连接对象、key前缀、有效期等
"""
@wraps(function)
def wrapper(*args, _save_engine=None, _key_prefix='cf', _expire_second=30, **kwargs):
"""
注意:
1. 如果函数执行时间过长，如果使用redis连接可能会断开
2. 注意kwargs中的参数不能和下面三个参数冲突
3. 缓存保存引擎只需要支持两个接口：
(1) get(key) -> value: 根据key获取获取的值
(2) set(key, value, ex=seconds): 保存缓存并设置有效期
:param _save_engine 缓存保存引擎，如redis.Redis()，如果该值为None则不进行缓存
:param _key_prefix str 缓存key前缀
:param _expire_second int 过期时间，单位：秒
"""
if _save_engine is None:
return function(*args, **kwargs)
key = "%s:%s:%s" % (_key_prefix, function.__name__, md5(str(args) + str(kwargs)[8:24]))
data = _save_engine.get(key)
if data:
return json.loads(data[1:])
data = function(*args, **kwargs)
_save_engine.set(key, "-"+json.dumps(data), ex=_expire_second)
return data
return wrapper其实，实现并没有强依赖于redis，只需要实现对应的接口即可。
参数名容易冲突的问题，给缓存控制的参数名加了下划线开头（这并不是完全解决冲突的问题，只是我们命名函数参数通常是不会以下划线开头的）。
而第三个问题的解决也很简单，只需要在缓存内容的前面增加一个字符即可。单元测试也可以模拟类似的情况：
def test_CacheFunc_None():
"""测试缓存"""
@CacheFunc
def func():
time.sleep(2)
return None
start = time.time()
res = func(_save_engine=redis_conn, _key_prefix='test3', _expire_second=6)
assert time.time() - start > 1
assert res is None
# 这时已经有缓存了
start = time.time()
res = func(_save_engine=redis_conn, _key_prefix='test3', _expire_second=6)
assert time.time() - start < 1
assert res is None这个实现已经很满意，也很通用了。
腾讯云开发者
