Python实现操作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")

来源：https://juejin.cn/post/7232253274057441317