Python实现操作Redis的高级用法分享

更新时间：2023年05月21日 11:07:25 作者：CodeDevMaster

redis-py是Python操作Redis的第三方库，它提供了与Redis服务器交互的API，本文为大家介绍了Python利用redis-py操作Redis的高级用法，需要的可以收藏一下

redis-py

redis-py是Python操作Redis的第三方库，它提供了与Redis服务器交互的API。

GitHub地址：https://github.com/redis/redis-py

安装redis-py

pip install redis

基本使用

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.Redis(host='localhost', port=6379, db=0)
# 设置键值对
r.set('hello', 'world')
# 获取键对应的值
value = r.get('hello')
# 输出 b'world'
print(value)
# 批量设置键值对
r.mset({'foo': '1', 'bar': '2'})
# 批量获取键对应的值
values = r.mget(['foo', 'bar'])
# 输出 [b'1', b'2']
print(values)

也可以使用StrictRedis对象连接redis服务器，StrictRedis类基于Redis类实现。

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 设置键值对
r.set('hello', 'world')
# 获取键对应的值
value = r.get('hello')
print(value) # 输出 b'world'

增删改查

以操作String数据类型的增删改查为例。

# 引⼊模块
import redis
if __name__ == '__main__':
    try:
        # 创建Redis对象
        r = redis.Redis(host='localhost', port=6379, db=0)
        # 新增，添加成功则返回True，如果添加失败则返回False
        result = r.set('name', 'test')
        print('是否新增成功：', result)
        # 获取，如果键存在则返回对应的值，如果键不存在则返回None
        name = r.get('name')
        print('查询结果：', name)
        # 修改，如果键已经存在则进⾏修改，如果键不存在则进⾏添加
        result = r.set('name', 'redis')
        print('是否修改成功：', result)
        name = r.get('name')
        print('查询结果：', name)
		# 获取所有的键
        result = r.keys()
        print('获取所有的键', result)
        # 删除，删除键及对应的值，如果删除成功则返回受影响的键数，否则则返 回0
        result = r.delete('name')
        print('删除key的数量：', result)
    except Exception as e:
        print(e)

字符串操作

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 设置键值对，键为'foo'，值为'bar'
r.set('foo', 'bar')
# 获取键对应的值
value = r.get('foo')
print(value) # 输出 b'bar'
# 批量设置键值对
r.mset({'apple': 'red', 'banana': 'yellow'})
# 批量获取键对应的值
values = r.mget(['apple', 'banana'])
print(values) # 输出 [b'red', b'yellow']
# 获取部分值
part_value = r.getrange('foo', 0, 1)
print(part_value) # 输出 b'ba'
# 追加字符串
r.append('foo', 'baz')
append_value = r.get('foo')
print(append_value) # 输出 b'barbaz'
# 自增计数器
r.incr('counter')
# 获取计数器的值
value = r.get('counter')
print(value) # 输出 b'1'
# 在自增计数器的基础上再加上5
r.incrby('counter', 5)
# 获取计数器的值
value = r.get('counter')
print(value) # 输出 b'6'
# 减少计数器
r.decr('counter')
# 获取计数器的值
value = r.get('counter')
print(value) # 输出 b'5'

哈希操作

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 设置哈希表键值对
r.hset('user1', 'name', 'Alice')
r.hset('user1', 'age', 20)
r.hset('user1', 'gender', 'female')
# 获取整个哈希表
hash_table = r.hgetall('user1')
print(hash_table) # 输出 {b'name': b'Alice', b'age': b'20', b'gender': b'female'}
# 获取特定键对应的值
value = r.hget('user1', 'name')
print(value) # 输出 b'Alice'
# 删除哈希表的一个键值对
r.hdel('user1', 'gender')
# 获取所有键名
keys = r.hkeys('user1')
print(keys) # 输出 [b'name', b'age']
# 获取所有键名对应的值
values = r.hvals('user1')
print(values) # 输出 [b'Alice', b'20']
# 批量设置哈希表键值对
r.hmset('user2', {'name': 'Bob', 'age': 25})
# 批量获取哈希表键名对应的值
values = r.hmget('user2', ['name', 'age'])
print(values) # 输出 [b'Bob', b'25']

列表操作

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 从左侧插入元素
r.lpush('mylist1', 'foo')
r.lpush('mylist1', 'bar')
r.lpush('mylist1', 'baz')
# 从右侧删除元素
r.rpop('mylist1')
# 获取列表长度
length = r.llen('mylist1')
print(length) # 输出 2
# 获取整个列表
mylist = r.lrange('mylist1', 0, -1)
print(mylist) # 输出 [b'baz', b'bar']
# 从左侧插入元素
r.lpush('mylist2', 'one')
r.lpush('mylist2', 'two')
r.lpush('mylist2', 'three')
# 弹出列表头部元素
value1 = r.lpop('mylist2')
print(value1) # 输出 b'three'
# 弹出列表尾部元素
value2 = r.rpop('mylist2')
print(value2) # 输出 b'one'
# 在指定元素前或后插入新元素
r.linsert('mylist2', 'BEFORE', 'two', 'new')
mylist = r.lrange('mylist2', 0, -1)
print(mylist) # 输出 [b'new', b'two']
# 裁剪列表
r.ltrim('mylist2', 0, 0)
mylist = r.lrange('mylist2', 0, -1)
print(mylist) # 输出 [b'new']

集合操作

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 向集合中添加元素
r.sadd('set1', 'foo')
r.sadd('set1', 'bar')
# 获取集合中的所有元素
members = r.smembers('set1')
print(members) # 输出 {b'foo', b'bar'}
# 获取集合中的元素数量
count = r.scard('set1')
print(count) # 输出 2
# 判断一个元素是否在集合中
result = r.sismember('set1', 'foo')
print(result) # 输出 True
# 删除集合中的一个元素
r.srem('set1', 'bar')
# 获取多个集合的交集
r.sadd('set2', 'foo')
r.sadd('set2', 'baz')
intersection = r.sinter(['set1', 'set2'])
print(intersection) # 输出 {b'foo'}
# 获取多个集合的并集
union = r.sunion(['set1', 'set2'])
print(union) # 输出 {b'foo', b'baz'}
# 获取一个集合与多个集合的差集
difference = r.sdiff('set2', ['set1'])
print(difference) # 输出 {b'baz'}

有序集合操作

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 添加有序集合成员和分值
r.zadd('zset1', {'foo': 1.0, 'bar': 2.0, 'baz': 4.0})
# 获取有序集合的成员数
count = r.zcard('zset1')
print(count) # 输出 3
# 获取有序集合指定范围内的成员
members = r.zrange('zset1', 0, -1)
print(members) # 输出 [b'foo', b'bar', b'baz']
# 获取有序集合指定成员的分值
score = r.zscore('zset1', 'bar')
print(score) # 输出 2.0
# 获取有序集合指定范围内成员的数量
count = r.zcount('zset1', 1.5, 3.5)
print(count) # 输出 1
# 删除有序集合中一个成员
r.zrem('zset1', 'bar')
# 获取有序集合中指定范围内的成员和分值
with_scores = r.zrangebyscore('zset1', 0, 5, withscores=True)
print(with_scores) # 输出 [(b'foo', 1.0), (b'baz', 4.0)]

高级用法

Redis管道pipeline

在Redis中，管道（pipeline）是指可以将多个Redis命令依次发送给Redis，让Redis 一次性执行这些命令并返回结果的机制。使用管道可以大大减少客户端与Redis的网络通信次数，提高Redis的处理效率，是优化Redis性能的重要手段之一。

在redis-py库中，可以使用pipeline()方法创建一个管道对象，并对该对象连续调用多个 Redis 命令并提交到 Redis 进行执行。提交执行后，每个命令都会获取到这些命令的执行结果，并按照请求的顺序返回给客户端。

特点：

1.可以一次性发送多条命令并在执行完后一次性将结果返回

2.pipeline通过减少客户端与Redis的通信次数来实现降低往返延时时间

实现原理：

管道pipeline实现的原理是队列，队列是先进先出，这样就保证数据的顺序性
Client可以将三个命令放到一个tcp报文一起发送
Server则可以将三条命令的处理结果放到一个tcp报文返回

基本使用

1.使用 pipeline() 方法创建一个新的 Pipeline 对象，并向该管道对象连续调用了三个不同的 SET 命令，分别设置了三个不同的键名和对应的键值

2.通过 execute() 方法提交管道内所有的命令。Redis 服务器一次性执行管道内所有的命令，并将结果返回给客户端

3.最后输出Redis 管道执行的结果到控制台，其中包含了每个 SET 命令的执行结果

import redis
# 连接到本地 Redis 服务器，默认端口为 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 使用 Redis 管道
pipe = r.pipeline()
# 连续设置多个键名和键值
pipe.set('name', 'John')
pipe.set('age', 30)
pipe.set('city', 'New York')
# 执行 Redis 管道中所有命令，并获取所有命令的执行结果
result = pipe.execute()
# 输出 Redis 管道执行结果
print(result)

Redis事务

事务 Redis 通过 MULTI 和 EXEC 来实现事务，MULTI 开启一个事务，EXEC 提交多个命令到 Redis 执行，可以保证单位时间内只有当前请求在访问 Redis 服务器，其他读写操作会等待这个事务结束后才能进行，从而保证了数据一致性。

基本使用

1.创建一个 redis.StrictRedis 实例，并且设置 transaction 参数为 True，表示开启 Redis 事务

2.使用 pipeline() 方法创建一个新的 Pipeline 对象，并将其 transaction 参数设置为 True，表示这个 Pipeline 是用于 Redis 事务的

3.调用 multi() 方法开启Redis 事务，之后向两个不同的键名 foo 和 bar 分别设置了不同的字符串值

4.最后通过 execute() 方法提交事务，Redis 将一次性执行整个事务并返回每个命令的执行结果

import redis
# 连接到本地 Redis 服务器，默认端口为 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 使用 Redis 事务
trans = r.pipeline(transaction=True)
# 开启事务
trans.multi()
# 向两个键名分别设置不同的值
trans.set('foo', 'hello')
trans.set('bar', 'world')
# 在 Redis 事务中执行以上命令，并获取执行结果
result = trans.execute()
# 输出 Redis 事务执行结果
print(result)

事务与管道的区别

在Redis中，事务（transaction）和管道（pipeline）都是用于批量执行命令的方式，但二者有本质上的不同：

1.调用方式不同

使用事务时，需要先通过MULTI命令将客户端设置为事务模式，然后按照一定的顺序添加执行的多个命令，最后通过EXEC命令将操作提交到服务器执行。

使用管道时，则是对同一个连接对象上连续调用多个Redis命令并且在最后统一执行这些命令。

2.发送机制不同

Redis事务的逻辑单元可以确保所有被包含的命令“原子性”地执行，即要么全部执行成功完成，要么全部回滚；而Redis>运用管道的方法仅仅是优化传输，将多个命令打包发送到Redis服务节点，并在结果关闭时进行收集处理，以达到多个请求一次通信的目的。

3.回滚能力不同

Redis事务提交的过程中如果某个命令执行失败了，后面的命令则都不会再执行，已经执行过的命令不会回滚。当然在EXEC之前可以通过DISCARD命令清空已经放入到事务队列里面的命令；而管道机制暂时没有回滚的能力。

因此:

Redis管道是解决高性能I/O操作的手段，主要目的在于将多个命令打包，一次发出去避免了每次发送都占有一个网络通道

Redis事务适用于数据的批量修改，并期望原子性action。两种方案各有利弊，需要按照实际业务场景选择使用哪一种方式。

分布式锁

Redis通过 SETNX 和 EXPIRE 等命令实现分布式锁，可防止多个客户端同时修改同一资源。具体实现时，检查一个键是否存在，若不存在则对该键进行设置并获得锁；若已存在则等待。

import time
import uuid
import redis
# 连接到本地 Redis 服务器，默认端口为 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
def acquire_lock(conn, lockname, acquire_timeout=10, lock_timeout=10):
    """
    尝试获取锁，获取成功返回锁id；获取失败或者异常返回None
    :param conn: Redis连接对象
    :param lockname: 锁的名称
    :param acquire_timeout: 最大尝试获取锁的时间（seconds）
    :param lock_timeout: 锁的超时时间（seconds）
    :return 是否获取权益：锁的ID/None
    """
    identifier = str(uuid.uuid4())
    end_time = time.time() + acquire_timeout
    while time.time() < end_time:
        # 获取锁
        if conn.setnx(lockname, identifier):
            conn.expire(lockname, lock_timeout)
            return identifier
        # 防止死锁
        elif not conn.ttl(lockname):
            conn.expire(lockname, lock_timeout)
        time.sleep(0.001)
    return None
def release_lock(conn, lockname, identifier):
    """
    根据锁id释放锁，若锁不存在或者已经被其他持有者所释放则返回False；成功释放返回True
    :param conn: Redis连接对象
    :param lockname: 锁的名称
    :param identifier：锁的ID
    :return 是否释放：True/False
    """
    pipe = conn.pipeline(True)
    while True:
        try:
            # 开启事务
            pipe.watch(lockname)
            # 检查对应键是否还是要当前程序设置的值，以avoid误删别的客户端的锁
            if pipe.get(lockname).decode('utf-8') == identifier:
                # 删除锁
                pipe.multi()
                pipe.delete(lockname)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:
            pass
    return False
if __name__ == '__main__':
    lockname = " lock"
    id = acquire_lock(r, lockname)
    print(id)
    tag = release_lock(r, lockname, id)
    print(tag)

订阅和发布

Redis 通过 SUBSCRIBE 和 PUBLISH 命令实现类似消息队列的功能，可以实现多个进程（客户端）针对一个频道进行消息的监听和广播。

import redis
# 连接到本地 Redis 服务器，默认端口为 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 在 “频道” 上发布 “消息”
pub = r.publish('channel', 'Hello World')
# 返回发布/订阅对象
sub = r.pubsub()
# 使用此对象，可以订阅频道并收听发布的消息
sub.subscribe('channel')
for message in sub.listen():
    print(message)
    print(message['data'])

GeoHash

Redis通过GeoHash 实现了地理位置排序和搜索等功能。

import redis
# 连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 添加地理位置坐标
r.geoadd("locations", 126.6507114792, 45.7603340486, "Harbin")
r.geoadd("locations", 117.2028132333, 39.0879960175, "Beijing")
r.geoadd("locations", 121.5670484719, 38.9544883509, "Tianjin")
# 获取指定两点间的距离（以 km 为单位）
distance = r.geodist("locations", "Harbin", "Beijing", unit="km")
print(distance)  # 1072.1429
# 搜索指定范围内的地理位置（以 km 为单位），并按照距离从近到远排序
locations = r.georadiusbymember("locations", "Tianjin", 500, unit="km", withdist=True, sort="ASC")
print(locations)  # [[b'Tianjin', 0.0], [b'Beijing', 377.3833]]

redis-py-cluster

redis-py-cluster是Python中用于连接Redis集群的模块，支持对Redis集群中的所有节点进行Hash槽分配操作，提供了与redis-py相同的API接口，使用方法类似。

GitHub地址：https://github.com/Grokzen/redis-py-cluster

安装

pip install redis-py-cluster

基本使用

from rediscluster import RedisCluster
# Redis节点
startup_nodes = [
    {"host": "127.0.0.1", "port": "6379"},
    {"host": "127.0.0.1", "port": "7001"},
    {"host": "127.0.0.1", "port": "7002"}
]
# 创建RedisCluster对象
rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)
# 与Redis的基本交互
rc.set("foo", "bar")
value = rc.get("foo")
print(value)  # bar
# 哈希操作
rc.hset("user", "name", "Alice")
rc.hset("user", "age", 20)
hash_table = rc.hgetall("user")
print(hash_table)  # {'name': 'Alice', 'age': '20'}
# 列表操作
rc.lpush("mylist", "foo")
rc.lpush("mylist", "bar")
rc.lpush("mylist", "baz")
mylist = rc.lrange("mylist", 0, -1)
print(mylist)  # ['baz', 'bar', 'foo']
# 删除键
rc.delete("foo")

以上就是Python实现操作Redis的高级用法分享的详细内容，更多关于Python操作Redis的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

Python中微服务架构的设计与实现详解
在当今软件开发领域中,微服务架构已经成为了一种流行的设计范式,这篇文章主要为大家介绍了如何使用Python语言来设计和实现微服务架构,需要的可以参考一下
2024-04-04
vscode autopep8无法格式化python代码问题解决
这篇文章主要为大家介绍了vscode autopep8无法格式化python代码问题解决,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
2023-09-09
Tensorflow tf.nn.atrous_conv2d如何实现空洞卷积的
这篇文章主要介绍了Tensorflow tf.nn.atrous_conv2d如何实现空洞卷积的，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-04-04
使用python+requests+pytest实现接口自动化
这篇文章主要介绍了使用python+requests+pytest实现接口自动化,在当前互联网产品迭代频繁的背景下，回归测试的时间越来越少，但接口自动化测试因其实现简单、维护成本低，容易提高覆盖率等特点，越来越受重视,需要的朋友可以参考下
2023-08-08
python @classmethod 的使用场合详解
这篇文章主要介绍了python @classmethod 的使用场合详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-08-08
Python 实现取多维数组第n维的前几位
今天小编就为大家分享一篇Python 实现取多维数组第n维的前几位，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-11-11
Python全局变量用法实例分析
这篇文章主要介绍了Python全局变量用法,结合实例形式分析了Python中全局变量的定义、使用方法与相关注意事项,需要的朋友可以参考下
2016-07-07
Python虚拟环境迁移的实现
本文主要介绍了Python虚拟环境迁移的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-03-03
Dockerfile构建一个Python Flask 镜像
这篇文章主要介绍了Dockerfile构建一个Python Flask 镜像，对正在学习的你有一定的参考价值，需要的小伙伴可以参考一下
2022-01-01
Selenium+Python自动化脚本环境搭建的全过程
说到自动化测试,就不得不提大名鼎鼎的Selenium,Selenium 是如今最常用的自动化测试工具之一,支持快速开发自动化测试框架,且支持在多种浏览器上执行测试,下面这篇文章主要给大家介绍了关于Selenium+Python自动化脚本环境搭建的相关资料,需要的朋友可以参考下
2021-09-09