测试平台系列(85) 把redis运用到实战中

米洛

教程

米洛的测开日记

回顾

上一节我们让支持了

前置条件

但在此之前，我想了一个很

严重

配置

我们的测试平台，后续会接入yapi，接入其他系统。势必会有一个地方去维护这些数据。

包括发件人邮箱，密码等等数据。

但这些数据又通常是全局共享，如果放到db的话，很鸡肋，因为数据只有1条，如果放到redis，有可能数据会丢。

博主也不知道放哪里比较好，最后决定放到一个configuration.json的配置文件里面了。

但是频繁读取文件，总归是不好的。而且我们线上会有许多个worker，还可能会有

冲突

想到我们之前拿捏过的redis，这不正是它的用武之地吗？

编写通用cache方法

在此之前，我们先思考一下

为啥要写这样的通用缓存办法

我们获取数据，有2部分，分别为get和set。结合缓存来看，我们可以写出这样的伪代码:

def get_cache():
data = redis.get(key)
if data is not None:
return data
data = get_data()
redis.set(key, data)
return data

就是这么简单的用法，如果key获取到了，我们直接return，如果没获取到，我们更新数据，并把数据写入redis，最后返回data。

那我们修改数据的时候怎么做呢？

def update_cache():
update(data)
redis.delete(key)

1. 先更新数据源
2. 删除缓存数据，这样下去获取缓存的时候就会

   重新获取数据

   并写入缓存

但大家有没有觉得这个过程很繁琐，而且属于

get和set之外

装饰器

编写cache装饰器

• 连接本地redis的方法

  首先我们在config.py配置好redis的连接信息，接着编写client客户端，因为它本身是连接池模式，所以我们一直用这个

  客户端

  都没问题。（所以我这里把它设置为了property）

• 编写RedisHelper

  helper类含有2个装饰器，cache负责读取(get)，up_cache负责更新(set)。

class RedisHelper(object):
pity_prefix = "pity"
pity_redis_client = PityRedisManager().client

@staticmethod
def get_key(key: str):
return f"{RedisHelper.pity_prefix}:{key}"

@staticmethod
def cache(key: str, expired_time=3 * 60):
"""
自动缓存装饰器
:param key: 被缓存的key
:param expired_time: 默认key过期时间
:return:
"""

def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
redis_key = RedisHelper.get_key(key)
data = RedisHelper.pity_redis_client.get(redis_key)
# 缓存已存在
if data is not None:
return json.loads(data)
# 获取最新数据
new_data = func(*args, **kwargs)
info = json.dumps(new_data)
RedisHelper.pity_redis_client.set(redis_key, info, ex=expired_time)
return new_data

return wrapper

return decorator

@staticmethod
def up_cache(key: str):
"""
redis缓存key，套了此方法，会自动执行更新数据操作后删除缓存
:param key:
:return:
"""

def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
redis_key = RedisHelper.get_key(key)
# 获取最新数据
new_data = func(*args, **kwargs)
# 更新数据，删除缓存
RedisHelper.pity_redis_client.delete(redis_key)
return new_data

return wrapper

return decorator

这里我们基本上按照之前说的

逻辑

编写配置文件获取方法

我们编写configuration.json到根目录:

import json
import os

from app.middleware.RedisManager import RedisHelper
from config import Config

class SystemConfiguration(object):
"""
系统配置
"""

@staticmethod
@RedisHelper.cache("configuration", 24 * 3600)
def get_config():
try:
filepath = os.path.join(Config.ROOT, "configuration.json")
if not os.path.exists(filepath):
raise Exception("没找到配置文件，请检查configuration文件是否已经被删除")
with open(filepath, mode="r", encoding='utf-8') as f:
return json.load(f)
except Exception as e:
raise Exception(f"获取系统设置失败, {e}")

@staticmethod
@RedisHelper.up_cache("configuration")
def update_config(config):
try:
filepath = os.path.join(Config.ROOT, "configuration.json")
if not os.path.exists(filepath):
raise Exception("没找到配置文件，请检查configuration文件是否已经被删除")
with open(filepath, mode="r", encoding='utf-8') as f:
json.dump(config, f)
except Exception as e:
raise Exception(f"更新系统设置失败, {e}")

由于配置文件一般很少更新，所以我们把key的过期时间设为了1天（其实可以更久一点）。

这样，我们调用get_config就可以拿到系统设置啦，里面有咱们很重要的发件人信息。

测试一下

启动程序以后，我们去查询redis中关于configuration的key，就用咱们自己写的客户端:

再测试下过期时间:

今天的内容就到这里，下节正式开启发邮件(报告通知)之旅。