flask插件系列之flask_caching缓存

Posted 2021-01-08 cwp-bg

前言

为了尽量减少缓存穿透，同时减少web的响应时间，我们可以针对那些需要一定时间才能获取结果的函数和那些不需要频繁更新的视图函数提供缓存服务，可以在一定的时间内直接返回结果而不是每次都需要计算或者从数据库中查找。flask_caching插件就是提供这种功能的神器。

flask_caching

安装

pip install Flask-Caching

初始化配置

# __init__.py
from flask import Flask
from extensions import cache
from setting import Config
app = Flask(__name__)
app.config.from_object(Config)
# 可以使用config参数添加配置
cache.init_app(app=app,config={‘CACHE_TYPE‘ ： ‘simple‘})

# extensions.py
from flask_caching import Cache

cache = Cache()

# 也可以使用配置文件的方式，推荐使用这种方式
# setting.py
class Config:
    CACHE_TYPE = ‘simple‘

常用配置参数详解

CACHE_TYPE:设置缓存的类型

# 下面五个参数是所有的类型共有的
CACHE_NO_NULL_WARNING = "warning" # null类型时的警告消息
CACHE_ARGS = [] # 在缓存类实例化过程中解包和传递的可选列表，即手动给缓存对象一个容器
CACHE_OPTIONS = {}  # 可选字典,在缓存类实例化期间传递
CACHE_DEFAULT_TIMEOUT # 默认过期/超时时间，单位为秒
CACHE_THRESHOLD # 缓存的最大条目数

CACHE_TYPE = null # 默认的缓存类型，无缓存
CACHE_TYPE = ‘simple‘ # 使用本地python字典进行存储，线程非安全

CACHE_TYPE = ‘filesystem‘ # 使用文件系统来存储缓存的值
CACHE_DIR = "" # 文件目录

CACHE_TYPE = ‘memcached‘ # 使用memcached服务器缓存
CACHE_KEY_PREFIX # 设置cache_key的前缀
CAHCE_MEMCACHED_SERVERS # 服务器地址的列表或元组
CACHE_MEMCACHED_USERNAME # 用户名
CACHE_MEMCACHED_PASSWORD # 密码

CACHE_TYPE = ‘uwsgi‘ # 使用uwsgi服务器作为缓存
CACHE_UWSGI_NAME # 要连接的uwsgi缓存实例的名称

CACHE_TYPE = ‘redis‘ # 使用redis作为缓存
CACHE_KEY_PREFIX # 设置cache_key的前缀
CACHE_REDIS_HOST  # redis地址
CACHE_REDIS_PORT  # redis端口
CACHE_REDIS_PASSWORD # redis密码
CACHE_REDIS_DB # 使用哪个数据库
# 也可以一键配置
CACHE_REDIS_URL 连接到Redis服务器的URL。示例redis://user:[email protected]:6379/2

使用实例

• 缓存函数或视图函数

# __init__.py
from flask import Flask, current_app, make_response, request
from test import get_result, test_xxx
from exsentions import cache

app = Flask(__name__)
app.config[‘SECRET_KEY‘] = ‘123‘
cache.init_app(app, config={ ‘CACHE_TYPE‘ : ‘redis‘,‘CACHE_REDIS_HOST‘:‘192.168.1.20‘,
                             ‘CACHE_REDIS_PORT‘:‘6390‘})

@app.route(‘/test1‘)
def test():
    form_dict = request.args
    n = form_dict.get(‘n‘)
    print(n)
    res = get_result(int(n))
    return res

@app.route(‘/test2‘)
@cache.cached(timeout=30) # 专门为装饰视图函数准备的装饰器
def test2():
    cache.delete_memoized(get_result,4)
    return ‘ok‘

@app.route(‘/test3‘)
def test3():
    res = test_xxx()
    return res


if __name__ == ‘__main__‘:
    app.run(host=‘127.0.0.1‘, port=80, debug=True)

# extensions.py
from flask_caching import Cache

cache = Cache()

# test.py
from exsentions import cache

@cache.memoize(timeout=60, make_name=‘get_result‘)
def get_result(n):
    return str(n)

@cache.cached(timeout=30)
def test_xxx():
    return ‘this is a test‘

• 缓存键值对

# __init__.py
from flask import Flask, current_app, make_response, request
from test import get_result, test_xxx
from exsentions import cache

app = Flask(__name__)
app.config[‘SECRET_KEY‘] = ‘123‘
cache.init_app(app, config={ ‘CACHE_TYPE‘ : ‘redis‘,‘CACHE_REDIS_HOST‘:‘192.168.1.20‘,
                             ‘CACHE_REDIS_PORT‘:‘6390‘})

@app.route(‘/test1‘)
def test():
    cache.set(‘name‘,‘xiaoming‘, timeout=30)
    cache.set(‘person‘,{‘name‘:‘aaa‘, ‘age‘:20})
    x = cache.get(‘name‘)
    print(x)
    cache.set_many([(‘name1‘,‘hhh‘),(‘name2‘,‘jjj‘)])
    print(cache.get_many("name1","name2"))
    print(cache.delete("name"))
    print(cache.delete_many("name1","name2"))
    return res

if __name__ == ‘__main__‘:
    app.run(host=‘127.0.0.1‘, port=80, debug=True)

分析

• Cache对象

Cache作为缓存对象，主要有三个参数：
app：app对象
with_jinja2_ext：是否支持jinja2语法，默认True
config:配置参数

• 主要方法

cache.cached：装饰器，装饰无参数函数，使得该函数结果可以缓存
参数:
timeout:超时时间
key_prefix：设置该函数的标志
unless：设置是否启用缓存，如果为True，不启用缓存
forced_update：设置缓存是否实时更新，如果为True，无论是否过期都将更新缓存
query_string：为True时，缓存键是先将参数排序然后哈希的结果

cache.memoize：装饰器，装饰有参数函数，使得该函数结果可以缓存
make_name：设置函数的标志，如果没有就使用装饰的函数
# 其他参数同cached

cache.delete_memoized：删除缓存
参数：
fname：缓存函数的名字或引用
*args：函数参数

cache.clear() # 清除缓存
cache.cache # 获取缓存对象

• BaseCache

所有类型的缓存对象都继承自flask框架的BaseCache,它定义了一个使用缓存的标准接口，通过继承BaseCache实现这些接口就可以在flask中方便的集成各种缓存；而flask_caching插件为我们对redis、memcached等实现了接口，做好了适配工作，我们可以直接使用。

from werkzeug.contrib.cache import BaseCache

# 标准接口
clear():清除缓存
get(key):获取一个键的值，如果是json格式会自动转化成字典
set(key,value,timeout):设置一个键值，value可以是字典，会自动转化json格式
set_many(key,value,timeout):设置多个键值对
add(key, value, timeout=None):设置一个键值,如果存在就pass
delete(key)：删除键
delete_many():删除多个键值
get_many():获取多个键的值
get_dict():获取多个键的值,返回一个字典
has():查询是否存在一个键
clear(self):清除缓存
inc(self, key, delta=1):将键的值加一
dec(self, key, delta=1)：将键的值减一

自定义缓存后端

如果flask_caching当前集成的缓存服务器不符合我们的要求，比如需要使用mongodb做为后端缓存，那么我们可以自己实现那些标准接口。

# mycache.py
# 定义缓存类
from werkzeug.contrib.cache import BaseCache

class MongoCache(BaseCache):
    def __init__(self, host=‘localhost‘, port=27017, username=None, password=None, default_timeout=500,**kwargs):
        BaseCache.__init__(self, default_timeout)
        if isinstance(host, str):
            try:
                from pymongo import MongoClient
            except ImportError:
                raise RuntimeError(‘no pymongo module found‘)
            self._client = MongoClient(host=host,port=port, username=username, password=password, **kwargs)
        else:
            self._client = host

    # 接下来使用pymongo实现BaseCache的各个接口


def mongo(app, config, args, kwargs):
    """
    这里处理app传进来的参数用来连接mongodb
    :param app:
    :param config:
    :param args:
    :param kwargs:
    :return:
    """
    args.append(app.config[‘MONGO_SERVERS‘])
    return MongoCache(*args, **kwargs)

# 初始化时
# __init__.py
from flask import Flask, current_app, make_response, request
from exsentions import cache

app = Flask(__name__)
app.config[‘SECRET_KEY‘] = ‘123‘
cache.init_app(app, config={ ‘CACHE_TYPE‘ : ‘mycache.mongo‘,
                            ‘MONGO_SERVERS‘:‘mongodb://username:[email protected]:27017/db‘})

• 参考：

• https://pythonhosted.org/Flask-Caching/#

• 作者：天宇之游
• 出处：http://www.cnblogs.com/cwp-bg/
• 本文版权归作者和博客园共有，欢迎转载、交流，但未经作者同意必须保留此段声明，且在文章明显位置给出原文链接。