redis 连接池

Posted 2020-10-17

redis是一个key-value存储系统，和memcached类似，支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set-有序集合)和hash(哈希类型)。这些数据类型都支持push／pop、add/remove及取交集和差集及更丰富的操作，而且这些操作都说原子性的。在此基础上，redis支持各种不同的方式排序。与memcached一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave（主从）同步。

 

python 操作redis，未使用连接池：

redis-py提供两个Redis和StrictRedis用于实现Redis的命令，StrictRedis用于实现大部分官方的命令，并使用官方的语法和命令，Redis是StrictRedis的子类，用于向后兼容旧版本的redis-py.

import redis
db = 0
#连接redis,password不简写(否则或报错)，db若不写，则默认操作db0
conn_redis = redis.Redis(host=\'127.0.0.1\', port=6379, password=\'123456\', db=db)
#给redis添加值，传值方式是key-value，key不可重复，value的形式尽量是string，也可以传list、字典，redis内存放的是字节res = conn_redis.set(\'name\', \'testredis\')
#print(res)    #返回值是布尔类型，set成功，则返回true
conn_redis.set(\'days\', \'[10,4,5,12,44]\')

#set key值到redis内，且可以设置过期时间，10s
result = conn_redis.setex(\'session\', \'abcder1233@sdfrr\', 10)   #单位是s
print(result)   #返回值是布尔类型，set成功，则返回true
#批量插入redis，可以写入多个key-valye
conn_redis.mset(a=\'xiaohei\', b=\'xiaobai\', c=\'xiaohuang\') 
#设置key时，可以设置文件夹，user文件夹，key:test,value:haha
conn_redis.set(\'user:test\',\'haha\')

python 操作redis，使用连接池：

redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。

def getcoon():
      pool = redis.ConnectionPool(host=\'192.168.1.112\', port=6379, password=\'123456\', db=0)
      coon = redis.Redis(connection_pool=pool)
      coon.set(\'key\', \'value\')
      res = coon.get(\'key\')
      return res

redis 使用连接池操作

class OPRedis(object):

    def __init__(self):
        if not hasattr(OPRedis, \'pool\'):
            OPRedis.getRedisCoon()  #创建redis连接
        self.coon = redis.Redis(connection_pool=OPRedis.pool)

    @staticmethod
    def getRedisCoon():
        OPRedis.pool = redis.ConnectionPool(host=redisInfo[\'host\'], password=redisInfo[\'password\'], port=redisInfo[\'port\'], db=redisInfo[\'db\'])

    """
    string类型 {\'key\':\'value\'} redis操作
    """

    def setredis(self, key, value, time=None):
        #非空即真非0即真
        if time:
            res = self.coon.setex(key, value, time)
        else:
            res = self.coon.set(key, value)
        return res

    def getRedis(self, key):
        res = self.coon.get(key).decode()
        return res

    def delRedis(self, key):
        res = self.coon.delete(key)
        return res

    """
    hash类型，{\'name\':{\'key\':\'value\'}} redis操作
    """
    def setHashRedis(self, name, key, value):
        res = self.coon.hset(name, key, value)
        return res


    def getHashRedis(self, name, key=None):
        # 判断key是否我为空，不为空，获取指定name内的某个key的value; 为空则获取name对应的所有value
        if key:
            res = self.coon.hget(name, key)
        else:
            res = self.coon.hgetall(name)
        return res


    def delHashRedis(self, name, key=None):
        if key:
            res = self.coon.hdel(name, key)
        else:
            res = self.coon.delete(name)
        return res

redisInfo配置

redisInfo = {
    "host": \'192.168.1.112\',
    "password": \'123456\',
    "port": 6379,
    "db": 0
}

创建test，测试redis连接

if __name__ == \'__main__\':
    opr = OPRedis()
    res = opr.setredis(\'two\', \'哈哈\')
    res = opr.getRedis(\'two\')
    print(res)

 

参考链接：https://www.cnblogs.com/xiaoming279/p/6293583.html