大家好~我是米洛！
我正在从0到1打造一个开源的接口测试平台, 也在编写一套与之对应的完整
教程，希望大家多多支持。
欢迎关注我的龚仲耗
测试开发坑货，获取最新文章教程!
回顾
上一节我们编写了Redis的相关配置编辑页面，博主这里也趁热打铁，把前端页面完善了。（可能会有一点点小问题，但应该主流程都正常）
其实和其他配置管理页面差不多，前端优化了一下面包屑，顶部的菜单也放回到左侧了。看看mac下的效果:
这里我给自己本地部署了个单实例的redis
搜索选项改动了一些，所见即所得，如果搜索项发生变化，那么内容也会随之切换
关于Redis客户端的选用
其实在这个问题上我是比较纠结的，redis有star很多的py客户端，也有与之对应的集群版本。但他们并不支持asyncio。
而支持asyncio的aioredis，本身是个很好的选择，但人家没有支持redis集群的计划。orz
所以今天想的是要不就用个同步的redis-cluster-py库算了，不过在我翻了github一段时间，发现了个叫aredis的异步库。大概瞅了下，他基本上是保持了和redis-cluster-py接近的api，可能也是为了吸引用户。
所以咱们就先试验一下，小白鼠嘛，总得有人来做。
安装aredis
看官网是要安装aredis[hiredis]，但我好像不适合这样方式，于是我分开装:
pip3 install aredis hiredis
编写RedisManager
其实这里还是和MySQL比较接近的，也是通过一个字典存放各个redis的连接配置。
不过由于Redis的集群和单实例还有一点区别（好在我们编写配置的时候就准备好了），所以我们最好是针对单实例和集群分别编写2个map存放他们的client，当然1个也是ok的。
整体流程: 从字典获取客户端，如果没有则新开一个客户端，并放入缓存，有则返回。
- 可能存在的问题
代码不是线程安全的，需要观察是否需要加锁
缓存不像LRU会降频，也不能自动过期
对我来说第一个肯定是个大问题，如果出现了就必须得解决。至于第二个问题，由于redis配置很少变动，而且我们本身是连接池的形式，所以影响不算大。
话不多说，现在我们就来编写吧:
""" redis客户端，基于aredis(支持集群，aioredis不支持集群) """ from aredis import StrictRedisCluster, ClusterConnectionPool, ConnectionPool, StrictRedis from app.excpetions.RedisException import RedisException class PityRedisManager(object): """非线程安全，可能存在问题 """ _cluster_pool = dict() _pool = dict() @staticmethod def get_cluster_client(redis_id: int, addr: str): """ 获取redis集群客户端 :param redis_id: :param addr: :return: """ cluster = PityRedisManager._cluster_pool.get(redis_id) if cluster is not None: return cluster client = PityRedisManager.get_cluster(addr) PityRedisManager._cluster_pool[redis_id] = client return client @staticmethod def get_single_node_client(redis_id: int, addr: str, password: str, db: str): """ 获取redis单实例客户端 :param redis_id: :param addr: :param password: :param db: :return: """ node = PityRedisManager._cluster_pool.get(redis_id) if node is not None: return node host, port = addr.split(":") pool = ConnectionPool(host=host, port=port, db=db, max_connections=100, password=password, decode_responses=True) client = StrictRedis(connection_pool=pool) PityRedisManager._pool[redis_id] = PityRedisManager.get_cluster(addr) return client @staticmethod def refresh_redis_client(redis_id: int, addr: str, password: str, db: str): """ 刷新redis客户端 :param redis_id: :param addr: :param password: :param db: :return: """ host, port = addr.split(":") pool = ConnectionPool(host=host, port=port, db=db, max_connections=100, password=password, decode_responses=True) client = StrictRedis(connection_pool=pool, decode_responses=True) PityRedisManager._pool[redis_id] = client @staticmethod def refresh_redis_cluster(redis_id: int, addr: str): PityRedisManager._cluster_pool[redis_id] = PityRedisManager.get_cluster(addr) @staticmethod def get_cluster(addr: str): """ 获取集群连接池 :param addr: :return: """ try: nodes = addr.split(',') startup_nodes = [{"host": n.split(":")[0], "port": n.split(":")[1]} for n in nodes] pool = ClusterConnectionPool(startup_nodes=startup_nodes, max_connections=100, decode_responses=True) client = StrictRedisCluster(connection_pool=pool, decode_responses=True) return client except Exception as e: raise RedisException(f"获取Redis连接失败, {e}")
我们以数据库的唯一id为key，缓存redis的连接池。
由于连接池会自动开启/关闭连接，所以我们不需要手动关闭客户端，非常方便。
仔细看看redis执行command的方法，里面会开辟连接，最终关闭连接，这就是连接池的好处，连接不会过期，因为每次都是新获取的
可以明显看到我们分别用了ClusterConnectionPool和ConnectionPool，分别对应集群和实例。参数基本上算是一致。
至于refresh，是给改动redis以后做的刷新连接的工作。
以上就是RedisManager的内容，到这只是能够获取Redis客户端了。
尝试一下
有条件的同学可以本次安装redis:
$ wget https://download.redis.io/releases/redis-6.2.6.tar.gz $ tar xzf redis-6.2.6.tar.gz $ cd redis-6.2.6 $ make
make了以后，修改redis-6.2.6目录下的redis.conf, 接着取消这一行的注释:
image
使用密码模式（redis最好是加密码，端口号也尽量不要用原生的6379，本宝宝有台机器被人通过redis植入了挖矿程序，苦不堪言）
- 在redis-6.2.6目录下启动
src/redis-server redis.conf
这样本地redis的实例就启动了~
image
编写个在线测试redis的接口
image
- 先通过id拿到redis的配置信息
- 然后通过manager拿到连接池
- 对redis发动命令
我们在http://localhost:7777/docs打开swagger调试: - 读取faker
image
- 设置faker为s12
image
- 再次取faker
image
可以看到redis的相关操作已经是可以用了，那我们今天的内容就到这了，愉快的周末总是辣么短暂。
下一节我们就得编写在线执行Redis的命令及相关页面了！
