简单封装Redis作缓存

基于Redis封装一个简单的Python缓存模块

0. Docker Redis安装

参考：

• Get Docker CE for CentOS
• 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客户端中a一、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. 待优化

如今又有问题了：

1. 如今直接用装饰器了，没有继承Cache类，如何设置过时时间或者标志位。
2. 怎么修饰类呢？形如这样的：

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

其余参考

• 详解Python的装饰器
• 可自定义属性的装饰器