基于Redis封装一个简单的Python缓存模块
0. Docker Redis安装
参考：
安装Docker时错误sudo yum-config-manager \ --add-repo \ https://download.docker.com/linux/centos/docker-ce.repo
因为配置没改，当时只改了yum的设置，再改次yum-config-manager的 vim /usr/bin/yum-config-manager-> !/usr/bin/python2.7(升级python时可参考CentOS7升级Python至2.7.13版本)
-
启动docker
sudo systemctl start docker -
拉取Redis镜像
docker pull redis -
启动Redis
docker run --name redis-master -p 6379:6379 -d redis -
查看容器情况
docker ps -
redis-cli 查看
docker exec -it 097efa63adef redis-cli
1. Python实例化Redis
import redis
class Cache(object):
def __init__(self):
pass
def _redis(self):
self._redis_pool = redis.ConnectionPool(host='127.0.0.1', port=6379)
return redis.Redis(connection_pool=self._redis_pool)
a = Cache()._redis()
a.set('a', '1')
a2 = Cache()._redis()
a2.set('b', '1')
a3 = Cache()._redis()
a3.set('c', '1')
for i in a.client_list():
print(i)
client_list()返回当前连接的客户端列表。
输出：
{'id': '73', 'addr': '127.0.0.1:54954', 'fd': '10', 'name': '', 'age': '0', 'idle': '0', 'flags': 'N', 'db': '0', 'sub': '0', 'psub': '0', 'multi': '-1', 'qbuf': '0', 'qbuf-free': '32768', 'obl': '0', 'oll': '0', 'omem': '0', 'events': 'r', 'cmd': 'client'}
{'id': '74', 'addr': '127.0.0.1:54955', 'fd': '8', 'name': '', 'age': '0', 'idle': '0', 'flags': 'N', 'db': '0', 'sub': '0', 'psub': '0', 'multi': '-1', 'qbuf': '0', 'qbuf-free': '32768', 'obl': '0', 'oll': '0', 'omem': '0', 'events': 'r', 'cmd': 'set'}
{'id': '75', 'addr': '127.0.0.1:54956', 'fd': '7', 'name': '', 'age': '0', 'idle': '0', 'flags': 'N', 'db': '0', 'sub': '0', 'psub': '0', 'multi': '-1', 'qbuf': '0', 'qbuf-free': '32768', 'obl': '0', 'oll': '0', 'omem': '0', 'events': 'r', 'cmd': 'set'}
多次连接会新开Redis占用资源，Redis用单例模式，直接新开文件，因为模块即单例。
将实例化Redis挪到新文件
import redis
_redis_pool = redis.ConnectionPool(host='127.0.0.1', port=6379)
_instance_redis = redis.Redis(connection_pool=_redis_pool)
注：
obj.__dict__输出所有对象时，连接池会出错，最后没用连接池：_instance_redis = redis.Redis(host='127.0.0.1', port=6379)
使用只需要引入，并在_redis()中返回实例即可
from Instance import _instance_redis
class Cache(object):
def __init__(self):
pass
def _redis(self):
return _instance_redis
a = Cache()._redis()
a.set('a', '1')
a2 = Cache()._redis()
a2.set('b', '1')
a3 = Cache()._redis()
a3.set('c', '1')
for i in a.client_list():
print(i)
输出：
{'id': '76', 'addr': '127.0.0.1:55070', 'fd': '11', 'name': '', 'age': '0', 'idle': '0', 'flags': 'N', 'db': '0', 'sub': '0', 'psub': '0', 'multi': '-1', 'qbuf': '0', 'qbuf-free': '32768', 'obl': '0', 'oll': '0', 'omem': '0', 'events': 'r', 'cmd': 'client'}
2. 生成Key
因为key手动指定，可能会指定一个已存在的Key，会覆盖其他记录。缓存需要随机Key。用uuid1生成随机ID。
import uuid
class Cache(object):
def key(self):
return uuid.uuid1()
但是这样用到缓存中是不行的，要做到相同的操作下得到的结果是相同的。所以可以将相同的操作这个动作转化成一个key存起来，下次再有此操作则使用缓存。比如get_all_store(status=1)函数在status==1时获取所有开业门店，status==2时获取所有打烊门店，就可有两个key分别做缓存。所以需要得到调用堆栈和最后的函数参数，分别哈希拼接起来即可做Key。（仅仅拿最后函数是不够的，之前的调用也需要拿到。因为函数可能不仅仅因为参数不一样数据结果就不一样，但调用堆栈相同情况下返回值应该是相同的。 比如同名函数和同名参数请求可能会重复。所以得有堆栈前缀。）
使用Python记录详细调用堆栈日志的方法的代码，获取到调用堆栈，单文件测试
F:\py\RedisCache> python .\Cache.py
Cache.py(<module>:103)->Cache.py(key:42)-> test
还算正常，从103行调用key()函数中(42行)的堆栈函数，参数为test
从其他文件引入Cache就坏了：
F:\py\RedisCache> python .\TestCache.py
TestCache.py(<module>:1)-><frozen importlib._bootstrap>(_find_and_load:983)-><frozen importlib._bootstrap>(_find_and_load_unlocked:967)-><frozen importlib._bootstrap>(_load_unlocked:677)-><frozen importlib._bootstrap_external>(exec_module:728)-><frozen importlib._bootstrap>(_call_with_frames_removed:219)->Cache.py(<module>:103)->Cache.py(key:42)-> test
TestCache.py(<module>:12)->Cache.py(key:42)-> test
输出了两次，第一次多了好多forzen importlib 开头的模块，第二次为想要的结果。
The module was found as a frozen module.
imp.PY_FROZEN
后来发现第一次输出那么多是因为TestCache调用Cache,但Cache文件中也有调用自身，加了__name__=='__main__'就好了。
修复一下，将每个过程拼接或者哈希拼接，然后:分隔。如果级别较多，只取三条即可，[-1] [-2]和之前的。等调用小节时候再说。
3. 基本操作
def set(self, key=None, value=None):
""" 设置缓存 """
if not key:
key = str(self.key())
self._redis.set(key, value, self._ttl)
def get(self, key):
""" 获取缓存 """
return self._redis.get(key)
def delete(self, key):
""" 删除缓存 """
return self._redis.delete(key)
def remove(self, pattern):
""" 批量删除缓存 """
del_count = 0
keys = self._redis.keys(pattern)
for key in keys:
if self.delete(key):
del_count += 1
return del_count
_ttl为在__init__()的设置过期时间，默认7200。其他属性还有：_redis()函数去掉，用_redis属性存储即可；_update和_delete请往后看。
def __init__(self):
# redis实例
self._redis = _instance_redis
# 过期时间
self._ttl = 7200
# 更新标志
self._update = False
# 删除标志
self._delete = False
那么，相应的就有设置属性操作：
def set_attr(self, **attr):
""" 设置属性 """
allows = ['update', 'delete', 'ttl']
for k in attr:
if k in allows:
name = str("_"+k)
setattr(self, name, attr[k])
设置属性示例：a1 用默认属性，a2用设置后的属性
c = Cache()
c.set("a1", 1)
print(c.__dict__)
c.set_attr(update=True, ttl=600).set('a2', 2)
print(c.__dict__)
查看：
# 程序输出：
{'_redis': Redis<ConnectionPool<Connection<host=127.0.0.1,port=6379,db=0>>>, '_ttl': 7200, '_update': False, '_delete': False}
{'_redis': Redis<ConnectionPool<Connection<host=127.0.0.1,port=6379,db=0>>>, '_ttl': 600, '_update': True, '_delete': False}
# 查看redis客户端中a1、a2键当前的剩余时间
127.0.0.1:6379> ttl a1
(integer) 7187
127.0.0.1:6379> ttl a2
(integer) 585
4. 定义调用缓存方法
第一版定义长这样：
def call(self, func):
""" 调用缓存方法 """
key = self.key()
cache = self.get(key)
# 删除缓存
if self._delete:
self._delete = True
return self.delete(key)
# 更新缓存
if not cache or self._update:
self._update = False
data = func()
value = json.dumps(data)
if self.set(key, value):
return data
return False
return json.loads(cache)
当设置更新_update或者无缓存时，执行函数更新缓存。
使用：
from Cache import Cache
class StoreCache(Cache):
""" 门店缓存类 """
def all_data(self, store_status):
""" 获取数据 """
def _(status=store_status):
print(f'func args status = {status}')
return [1, 2, 3, 4, 5]
return super().call(_)
if __name__ == '__main__':
s = StoreCache()
data = s.all_data(5)
print(data)
这样一看都是PHP的思想，用匿名函数做参数，传过去，需要定义缓存的时候执行函数。在Python用的时候，这样有弊端，函数参数得有默认值，最后生成Key获取参数也不方便。所以，改用装饰器。
def __init__(self, func=None):
# 获取缓存函数
self._cache_func = func
def __call__(self, *args, **kwargs):
""" 调用缓存方法 """
# 存储函数参数
self._cache_func.args = args
self._cache_func.kwargs = kwargs
# 获取key，取缓存
key = self.key()
cache = self.get(key)
# 删除缓存
if self._delete:
self._delete = True
return self.delete(key)
# 更新缓存
if not cache or self._update:
self._update = False
data = self._cache_func(*args, **kwargs)
value = json.dumps(data)
if self.set(key, value):
return data
return False
return json.loads(cache)
生成Key优化
生成Key时过滤前两次，第一次为key函数本身，第二次为__call__调用函数，都可忽略。
'Cache.py(key:35)', 'Cache.py(__call__:91)',
f = sys._getframe()
f = f.f_back # 第一次是key函数自身，忽略
f = f.f_back # 第二次是Cache文件的__call__，忽略
等价于f = sys._getframe(2)
Key最终生成规则：
使用有效的（除了最近两次缓存文件自身调用(key()和__call__())以及缓存函数以外）调用堆栈字符串（细化到函数名），堆栈哈希值（粒度细化到函数不需要堆栈哈希值或者说不需要细化到行号），缓存函数，参数哈希值（*args, **kwargs）:
有效堆栈字符串:缓存函数:参数哈希值
形如：
TestCache:func:xxxxxxxxxxxxxxxxxxxxx
5. 示例
5.1 函数缓存示例
新建文件TestCache.py
使用Cache装饰了all_data()方法，此方法下次使用会生成缓存。
from Cache import Cache
@Cache
def all_data(status):
""" 缓存数据缓存 """
print(f"this is all data, args={status}")
return list(range(status)) # range(status)用来生成模拟数据
class TestC(object):
def get(self):
t1 = all_data(10)
return t1
if __name__ == '__main__':
a1 = all_data(10)
print(a1)
a2 = all_data(10)
print(a2)
a3 = all_data(1)
print(a3)
a4 = TestC().get()
print(a4)
输出结果：a1首先写缓存，进入了函数，a2和a1Key相同，使用a1缓存。
a3参数不同，生成的Key也不同，写缓存。a4调用栈不同所以Key不同，写缓存。
F:\py\RedisCache> python .\TestCache.py
key:TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C
this is all data, args=10
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
key:TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
key:TestCache:<module>:all_data:B9E5D26E4217C1CB496844E233F59E17
this is all data, args=1
[0]
key:TestCache:<module>:TestCache:get:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C
this is all data, args=10
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
redis 输出
127.0.0.1:6379> keys *
1) "TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C"
2) "TestCache:<module>:all_data:B9E5D26E4217C1CB496844E233F59E17"
3) "TestCache:<module>:TestCache:get:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C"
如此，就可以装饰函数使用缓存了。
注：如果Redis的Key形如
2) "ptvsd_launcher:<module>:__main__:main:__main__:handle_args:_local:debug_main:_local:run_file:_local:_run:pydevd:main:pydevd:run:pydevd:_exec:_pydev_execfile:execfile:TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C"
3) "ptvsd_launcher:<module>:__main__:main:__main__:handle_args:_local:debug_main:_local:run_file:_local:_run:pydevd:main:pydevd:run:pydevd:_exec:_pydev_execfile:execfile:TestCache:<module>:all_data:B9E5D26E4217C1CB496844E233F59E17"
多了很多莫名其妙的调用，那是因为VsCode调试模式使用了ptvsd模块。
6. 待优化
现在又有问题了：
- 现在直接用装饰器了，没有继承Cache类，如何设置过期时间或者标志位。
- 怎么修饰类呢？形如这样的：
class Manager(object):
@Cache
def search_manager(self, district=1):
print("into search manager func.")
return list(range(district))
下周好好研究下装饰器再优化吧，别忘了弄这个缓存是为了给公众号添加功能的工具。
6.1 指定Key
发现有个必要的问题还得改：指定Key。不然Token没办法存，多个地方调用的不一样，必须有唯一Key。
需要指定Key的话，装饰器就要这么写了：
def __init__(self, key=None):
# 指定key
self._key = key
# 缓存函数
self._cache_func = None
# redis实例
self._redis = _instance_redis
# 过期时间
self._ttl = 7200
# 更新标志
self._update = False
# 删除标志
self._delete = False
def __call__(self, func):
""" 调用缓存 """
self._cache_func = func
def wrapper(*args, **kwargs):
# 存储函数参数
self._cache_func.args = args
self._cache_func.kwargs = kwargs
# 获取key，获取缓存
key = self.key()
cache = self.get(key)
# 删除缓存
if self._delete:
self._delete = True
return self.delete(key)
# 更新缓存
if not cache or self._update:
self._update = False
data = func(*args, **kwargs)
value = json.dumps(data)
if self.set(key, value):
return data
return False
return json.loads(cache)
return wrapper
def key(self):
""" 生成Key """
if self._key:
""" 使用指定Key """
key = self._key
logging.debug("key: %s" % key)
return key
......
调用时指定all_data_key函数有唯一Key：
from Cache import Cache
@Cache()
def all_data(status):
""" 缓存数据 """
print(f"this is all data, args={status}")
return list(range(status))
class TestC(object):
""" 类中调用查看堆栈不同 """
def get(self):
t1 = all_data(10)
return t1
@Cache(key='X0011')
def all_data_key(status):
""" 指定Key的缓存数据 """
print(f"this is all data (use key), args={status}")
return list(range(status))
if __name__ == '__main__':
a1 = all_data(10)
print(a1)
a2 = all_data(10)
print(a2)
a3 = all_data(1)
print(a3)
a4 = TestC().get()
print(a4)
print("use key: -------------------------- ")
a5 = all_data_key(4)
print(a5)
a6 = all_data_key(5)
print(a6)
输出：
DEBUG:root:key:TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C
this is all data, args=10
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
DEBUG:root:key:TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
DEBUG:root:key:TestCache:<module>:all_data:B9E5D26E4217C1CB496844E233F59E17
this is all data, args=1
[0]
DEBUG:root:key:TestCache:<module>:TestCache:get:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C
this is all data, args=10
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
use key: --------------------------
DEBUG:root:key: X0011
this is all data (use key), args=4
[0, 1, 2, 3]
DEBUG:root:key: X0011
[0, 1, 2, 3]
# redis查看
1) "TestCache:<module>:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C"
2) "TestCache:<module>:all_data:B9E5D26E4217C1CB496844E233F59E17"
3) "X0011"
4) "TestCache:<module>:TestCache:get:all_data:A66EEA7FC6E56FCAA27D26DF40CE5F2C"
可以看到all_data_key(4)和all_data_key(5)函数使用相同的Key X0011所以第二次函数未执行。 不受参数所限制，指定后到过期前就不能更换了，所以适用唯一的函数。当然了，根据业务权衡，是要缓存还是要Key。能满足各自的需求就可以了。不满足自己改规则。
6.2 优化类函数缓存
改了上面的装饰器之后，用
class Manager():
@Cache()
def search_manager(self, district=1):
print("into search manager func.")
return list(range(district))
可以跑通，但是key不唯一，因为传给__call__()的args参数形如(<__main__.Manager object at 0x030FDA50>, 3)所以每次都不一样。前面那一串<__main__.Manager object at 0x030FDA50>就是Manager实例化的self。过滤掉就可以了。
第一次尝试
args_temp = self._cache_func.args
if isinstance(args_temp[0], object):
args_temp = args_temp[1:]
这样可以，但是万一人第一个参数本来就是对象呢，岂不误删了。
再次尝试
不仅判断object，还要判断该函数是否属于self的类函数：
print(self._cache_func, type(self._cache_func))
print(self._cache_func.args[0], type(self._cache_func.args[0]))
# 分别输出：
<function Manager.search_manager at 0x03062F60> <class 'function'>
<__main__.Manager object at 0x03321050> <class '__main__.Manager'>
看来直接if isinstance(self._cache_func, type(self._cache_func.args[0])): 也是不行的，一个是函数对象，一个是类对象，统一拿到类名再比较：
# 过滤参数中的self对象
args_temp = self._cache_func.args
# 拿到类函数对象的类名和类对象的类名
func_class_name = os.path.splitext(self._cache_func.__qualname__)[0]
obj_class_name = self._cache_func.args[0].__class__.__name__
if isinstance(args_temp[0], object) and func_class_name == obj_class_name:
args_temp = args_temp[1:]
测试一波：
""" 类函数缓存测试 """
from Cache import Cache
class Manager():
""" 测试类函数缓存 """
@Cache()
def search_manager(self, district=1):
print("into search manager func.")
return list(range(district))
@Cache()
def search_manager(obj, district=1):
""" 测试对象过滤 """
print("into search manager func.")
return list(range(district))
if __name__ == '__main__':
a1 = Manager().search_manager()
print(a1)
a2 = Manager().search_manager(2)
print(a2)
a3 = Manager().search_manager(2)
print(a3)
print("test object: ---------------")
m1 = Manager()
m2 = Manager()
b1 = search_manager(m1, 2)
b2 = search_manager(m1, 2)
b3 = search_manager(m2, 2)
输出
DEBUG:root:key:TestClassCache:<module>:search_manager:E517FB10ADC90F5B727C5D734FD63EBC
into search manager func.
[0]
DEBUG:root:key:TestClassCache:<module>:search_manager:F575A889334789CA315DF7C855F33BEC
into search manager func.
[0, 1]
DEBUG:root:key:TestClassCache:<module>:search_manager:F575A889334789CA315DF7C855F33BEC
[0, 1]
test object: ---------------
DEBUG:root:key:TestClassCache:<module>:search_manager:933A83EC830CD986E4CA81EA3A9A260C
into search manager func.
DEBUG:root:key:TestClassCache:<module>:search_manager:933A83EC830CD986E4CA81EA3A9A260C
DEBUG:root:key:TestClassCache:<module>:search_manager:4183D446C799065E0DAD7FCC47D934C3
into search manager func.
最后的b1 b2同使用m1对象，b3使用m2对象，可以观察出来没有过滤此对象。
6.3 设置属性
不同的调用都应该可以设置所调用函数的生存时间，以及是否强制更新/删除。
失败尝试1：装饰器参数
开始想直接带参数即可。
在装饰时候加参数：像之前的key参数一样
@Cache(key='A001', ttl=100, update=True)
def attack_wait():
pass
在Cache初始化中设置：
class Cache(object):
def __init__(self, *, key=None, **attr):
pass
# 设置属性
if attr:
self.set_attr(**attr)
这样试了一下，不可以。
因为装饰器@Cache(xxxx)在代码运行到这里就会执行__init__，所以设置了三次在调用之前就会初始化装饰器
attr:
{}
attr:
{'ttl': 100}
attr:
{'ttl': 100, 'update': True}
达不到每次函数在多个地方不同的需求。
直接在函数参数里做手脚吧：
失败尝试2：缓存函数参数
直接在缓存里带约定好的参数
attack_start(cache_attr={'ttl':100, 'update':True})
还要在缓存函数中增加参数：
@Cache()
def attack_start(*, cache_attr=None):
在Cache中修改：
def __call__(self, func):
""" 调用缓存 """
self._cache_func = func
@functools.wraps(func)
def wrapper(*args, **kwargs):
if 'cache_attr' in kwargs:
print(kwargs['cache_attr'])
self.set_attr(**kwargs['cache_attr'])
可以完成但是太蠢了。
有没有不传参而且接触不到原对象但能在运行时影响原对象属性的？好像做不到。
可自定义属性的装饰器
直到我找到了9.5 可自定义属性的装饰器：
你想写一个装饰器来包装一个函数，并且允许用户提供参数在运行时控制装饰器行为。
添加装饰器
def attach_wrapper(obj, func=None):
if func is None:
return partial(attach_wrapper, obj)
setattr(obj, func.__name__, func)
return func
在__call__中就可以用函数了，因为之前有写set_attr()方法，直接调用一次就好了，精简后：
@attach_wrapper(wrapper)
def set_attr(**attr):
self.set_attr(**attr)
这样就好了。测试一下：
@Cache()
def attack_start(mul=1):
print("战斗开始......")
return 'attack ' * mul
if __name__ == "__main__":
attack_start.set_attr(update=True, ttl=200)
a1 = attack_start(3)
print(a1)
attack_start.set_attr(update=True, ttl=100)
a2 = attack_start(3)
print(a2)
attack_start.set_attr(delete=True)
a3 = attack_start(3)
print(a3)
输出：
DEBUG:root:key:TestCacheAttr:<module>:attack_start:CFDA50BC76FD8A05597004C3B00E927E
DEBUG:root:cache attr:{'_key': None, '_cache_func': <function attack_start at 0x038B07C8>, '_redis': Redis<ConnectionPool<Connection<host=127.0.0.1,port=6379,db=0>>>, '_ttl': 200, '_update': True, '_delete': False, '_attr': {}}
战斗开始......
DEBUG:root:redis set: TestCacheAttr:<module>:attack_start:CFDA50BC76FD8A05597004C3B00E927E:"attack attack attack ",(200s)
attack attack attack
DEBUG:root:key:TestCacheAttr:<module>:attack_start:CFDA50BC76FD8A05597004C3B00E927E
DEBUG:root:cache attr:{'_key': None, '_cache_func': <function attack_start at 0x038B07C8>, '_redis': Redis<ConnectionPool<Connection<host=127.0.0.1,port=6379,db=0>>>, '_ttl': 100, '_update': True, '_delete': False, '_attr': {}}
战斗开始......
DEBUG:root:redis set: TestCacheAttr:<module>:attack_start:CFDA50BC76FD8A05597004C3B00E927E:"attack attack attack ",(100s)
attack attack attack
DEBUG:root:key:TestCacheAttr:<module>:attack_start:CFDA50BC76FD8A05597004C3B00E927E
1
设置三次，前两次强制更新并设置ttl，所以前一二次虽然Key相同也可以更新，从ttl可以看出来200->100。第三次删除，查看Redis列表为空。
改了功能还不影响以前的代码逻辑，这就是装饰器的妙处吧。如此简单和圆满，美滋滋。
完整代码GitHub