美文网首页
测试平台系列(80) 封装Redis客户端

测试平台系列(80) 封装Redis客户端

作者: 米洛丶 | 来源:发表于2021-11-25 14:33 被阅读0次

    大家好~我是米洛

    我正在从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的命令及相关页面了!

    相关文章

      网友评论

          本文标题:测试平台系列(80) 封装Redis客户端

          本文链接:https://www.haomeiwen.com/subject/psnatrtx.html