Python操作Redis数据库入门教程

1. Redis 简介

Redis 是一个开源的（BSD 许可），基于内存的数据结构存储系统，它可以用作数据库、缓存和消息中间件。Redis 支持多种数据结构，例如字符串（strings）、哈希（hashes）、列表（lists）、集合（sets）、有序集合（sorted sets）等。由于其基于内存的特性，Redis 提供了非常高的读写性能，在高并发场景下表现出色，被广泛应用于 Web 应用程序、实时分析、缓存、队列等领域。

2. Python 与 Redis 的交互库

在 Python 中，有多个库可以与 Redis 进行交互，其中最常用的是 redis - py。redis - py 是 Redis 官方推荐的 Python 客户端库，它提供了简洁且功能丰富的 API 来操作 Redis 数据库。

安装 redis - py 非常简单，使用 pip 包管理器即可：

pip install redis

如果使用的是 pipenv 或 conda，相应的安装命令如下：

• pipenv install redis
• conda install -c conda - forge redis

3. 连接 Redis 数据库

在 Python 代码中，首先需要建立与 Redis 数据库的连接。redis - py 提供了 Redis 类来表示与 Redis 服务器的连接。以下是一个简单的连接示例：

import redis

# 创建 Redis 连接对象
r = redis.Redis(host='localhost', port=6379, db = 0)

# 测试连接
try:
    pong = r.ping()
    if pong:
        print('成功连接到 Redis 服务器')
except redis.exceptions.ConnectionError as e:
    print(f'连接 Redis 服务器失败: {e}')

在上述代码中：

• redis.Redis(host='localhost', port=6379, db = 0) 创建了一个 Redis 对象，指定了 Redis 服务器的主机为 localhost（如果 Redis 服务器在本地运行），端口为默认的 6379，并且选择了数据库 0。Redis 支持多个逻辑数据库，通过 db 参数指定要使用的数据库编号，编号从 0 开始。
• r.ping() 方法用于测试与 Redis 服务器的连接，成功连接时会返回 True，否则会抛出 redis.exceptions.ConnectionError 异常。

4. 操作字符串类型

Redis 中的字符串类型是最基本的数据类型，它可以存储任何类型的数据，例如文本、二进制数据等。在 redis - py 中，提供了一系列方法来操作字符串类型的数据。

4.1 设置字符串值

使用 set 方法可以设置一个键值对：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 设置键为 'name'，值为 'John'
result = r.set('name', 'John')
print(result)

在上述代码中，r.set('name', 'John') 将键 name 的值设置为 John。set 方法在设置成功时返回 True，否则返回 False。

4.2 获取字符串值

使用 get 方法可以获取指定键的值：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 获取键为 'name' 的值
name = r.get('name')
if name:
    print(name.decode('utf - 8'))

在上述代码中，r.get('name') 获取键 name 的值。由于从 Redis 中获取的数据是字节类型，所以需要使用 decode('utf - 8') 将其转换为字符串类型（前提是存储的数据是 UTF - 8 编码的文本）。

4.3 自增和自减

如果存储的字符串值是数字类型，还可以进行自增和自减操作。incr 方法用于自增，decr 方法用于自减：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 设置键为 'count'，初始值为 10
r.set('count', 10)

# 自增操作
new_count = r.incr('count')
print(new_count)

# 自减操作
new_count = r.decr('count')
print(new_count)

在上述代码中：

• r.incr('count') 将键 count 的值自增 1，并返回自增后的值。
• r.decr('count') 将键 count 的值自减 1，并返回自减后的值。

5. 操作哈希类型

哈希类型用于存储字段和值的映射关系，类似于 Python 中的字典。在 Redis 中，一个哈希可以包含多个字段，每个字段都有一个对应的值。

5.1 设置哈希值

使用 hset 方法可以设置哈希中的一个字段值，使用 hmset 方法可以一次性设置多个字段值：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 设置单个字段值
result1 = r.hset('user:1', 'name', 'Alice')

# 设置多个字段值
data = {
    'age': 25,
    'email': 'alice@example.com'
}
result2 = r.hmset('user:1', data)

print(result1)
print(result2)

在上述代码中：

• r.hset('user:1', 'name', 'Alice') 将哈希 user:1 中的 name 字段设置为 Alice。
• r.hmset('user:1', data) 将哈希 user:1 中的多个字段按照 data 字典中的内容进行设置。hset 和 hmset 方法在设置成功时返回 True，否则返回 False。

5.2 获取哈希值

使用 hget 方法可以获取哈希中一个字段的值，使用 hmget 方法可以获取多个字段的值，使用 hgetall 方法可以获取哈希中的所有字段和值：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 获取单个字段值
name = r.hget('user:1', 'name')
if name:
    print(name.decode('utf - 8'))

# 获取多个字段值
fields = ['age', 'email']
values = r.hmget('user:1', fields)
print([v.decode('utf - 8') if v else None for v in values])

# 获取所有字段和值
all_data = r.hgetall('user:1')
print({k.decode('utf - 8'): v.decode('utf - 8') for k, v in all_data.items()})

在上述代码中：

• r.hget('user:1', 'name') 获取哈希 user:1 中 name 字段的值，并进行解码。
• r.hmget('user:1', fields) 获取哈希 user:1 中 fields 列表指定的多个字段的值，并进行解码处理。
• r.hgetall('user:1') 获取哈希 user:1 中的所有字段和值，并将字节类型转换为字符串类型。

5.3 删除哈希字段

使用 hdel 方法可以删除哈希中的一个或多个字段：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 删除哈希中的 'email' 字段
result = r.hdel('user:1', 'email')
print(result)

在上述代码中，r.hdel('user:1', 'email') 删除哈希 user:1 中的 email 字段，删除成功返回被删除的字段数量。

6. 操作列表类型

Redis 中的列表是一个有序的字符串元素集合，支持在列表的两端进行插入和删除操作。

6.1 向列表中添加元素

使用 lpush 方法可以在列表的头部插入一个或多个元素，使用 rpush 方法可以在列表的尾部插入一个或多个元素：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 在列表 'fruits' 的头部插入元素
r.lpush('fruits', 'apple')
r.lpush('fruits', 'banana')

# 在列表 'fruits' 的尾部插入元素
r.rpush('fruits', 'cherry')

在上述代码中：

• r.lpush('fruits', 'apple') 在列表 fruits 的头部插入 apple 元素。
• r.rpush('fruits', 'cherry') 在列表 fruits 的尾部插入 cherry 元素。

6.2 从列表中获取元素

使用 lrange 方法可以获取列表中的一段元素：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 获取列表 'fruits' 中的所有元素
fruits = r.lrange('fruits', 0, -1)
print([fruit.decode('utf - 8') for fruit in fruits])

在上述代码中，r.lrange('fruits', 0, -1) 获取列表 fruits 中从索引 0 到最后一个元素（索引 -1 表示最后一个元素）的所有元素，并进行解码。

6.3 从列表中弹出元素

使用 lpop 方法可以从列表的头部弹出一个元素，使用 rpop 方法可以从列表的尾部弹出一个元素：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 从列表 'fruits' 的头部弹出一个元素
popped = r.lpop('fruits')
if popped:
    print(popped.decode('utf - 8'))

在上述代码中，r.lpop('fruits') 从列表 fruits 的头部弹出一个元素，并进行解码。

7. 操作集合类型

Redis 中的集合是一个无序的字符串元素集合，集合中的元素是唯一的，不允许重复。

7.1 向集合中添加元素

使用 sadd 方法可以向集合中添加一个或多个元素：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 向集合 'numbers' 中添加元素
r.sadd('numbers', 1)
r.sadd('numbers', 2, 3)

在上述代码中：

• r.sadd('numbers', 1) 向集合 numbers 中添加元素 1。
• r.sadd('numbers', 2, 3) 向集合 numbers 中添加元素 2 和 3。

7.2 获取集合中的所有元素

使用 smembers 方法可以获取集合中的所有元素：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 获取集合 'numbers' 中的所有元素
numbers = r.smembers('numbers')
print([number.decode('utf - 8') for number in numbers])

在上述代码中，r.smembers('numbers') 获取集合 numbers 中的所有元素，并进行解码。

7.3 判断元素是否在集合中

使用 sismember 方法可以判断一个元素是否在集合中：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 判断元素 2 是否在集合 'numbers' 中
is_member = r.sismember('numbers', 2)
print(is_member)

在上述代码中，r.sismember('numbers', 2) 判断元素 2 是否在集合 numbers 中，返回 True 或 False。

8. 操作有序集合类型

Redis 中的有序集合与集合类似，也是一个无序的字符串元素集合，但是每个元素都关联了一个分数（score），通过分数来对元素进行排序。

8.1 向有序集合中添加元素

使用 zadd 方法可以向有序集合中添加一个或多个元素，每个元素需要指定一个分数：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 向有序集合 'ranks' 中添加元素
r.zadd('ranks', {'Alice': 85, 'Bob': 90, 'Charlie': 78})

在上述代码中，r.zadd('ranks', {'Alice': 85, 'Bob': 90, 'Charlie': 78}) 向有序集合 ranks 中添加了三个元素，分别是 Alice（分数为 85）、Bob（分数为 90）和 Charlie（分数为 78）。

8.2 获取有序集合中的元素

使用 zrange 方法可以按照分数从小到大的顺序获取有序集合中的一段元素，使用 zrevrange 方法可以按照分数从大到小的顺序获取有序集合中的一段元素：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 按照分数从小到大获取有序集合 'ranks' 中的所有元素
ranks_asc = r.zrange('ranks', 0, -1, withscores=True)
print([(member.decode('utf - 8'), score) for member, score in ranks_asc])

# 按照分数从大到小获取有序集合 'ranks' 中的所有元素
ranks_desc = r.zrevrange('ranks', 0, -1, withscores=True)
print([(member.decode('utf - 8'), score) for member, score in ranks_desc])

在上述代码中：

• r.zrange('ranks', 0, -1, withscores=True) 按照分数从小到大获取有序集合 ranks 中的所有元素，并同时获取每个元素的分数。
• r.zrevrange('ranks', 0, -1, withscores=True) 按照分数从大到小获取有序集合 ranks 中的所有元素，并同时获取每个元素的分数。

8.3 获取有序集合中元素的排名

使用 zrank 方法可以获取元素在有序集合中的排名（从 0 开始，分数从小到大排序），使用 zrevrank 方法可以获取元素在有序集合中的排名（从 0 开始，分数从大到小排序）：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 获取元素 'Bob' 在有序集合 'ranks' 中的排名（分数从小到大）
rank_asc = r.zrank('ranks', 'Bob')
print(rank_asc)

# 获取元素 'Bob' 在有序集合 'ranks' 中的排名（分数从大到小）
rank_desc = r.zrevrank('ranks', 'Bob')
print(rank_desc)

在上述代码中：

• r.zrank('ranks', 'Bob') 获取元素 Bob 在有序集合 ranks 中按照分数从小到大排序的排名。
• r.zrevrank('ranks', 'Bob') 获取元素 Bob 在有序集合 ranks 中按照分数从大到小排序的排名。

9. 事务操作

Redis 支持事务操作，通过事务可以将多个命令组合在一起，保证这些命令要么全部执行成功，要么全部不执行。在 redis - py 中，可以使用 pipeline 对象来实现事务操作。

9.1 简单事务示例

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 创建管道对象
pipe = r.pipeline()

try:
    # 开始事务
    pipe.multi()

    # 执行多个命令
    pipe.set('key1', 'value1')
    pipe.set('key2', 'value2')

    # 执行事务
    pipe.execute()
    print('事务执行成功')
except redis.exceptions.RedisError as e:
    print(f'事务执行失败: {e}')
    # 回滚事务（在实际应用中可能需要根据具体情况处理回滚逻辑）
    pipe.reset()

在上述代码中：

• pipe = r.pipeline() 创建了一个管道对象。
• pipe.multi() 开始一个事务，之后在管道对象上执行的命令不会立即执行，而是被缓冲起来。
• pipe.set('key1', 'value1') 和 pipe.set('key2', 'value2') 是添加到事务中的命令。
• pipe.execute() 执行事务，将缓冲的命令一次性发送到 Redis 服务器执行。如果在事务执行过程中发生错误，execute 方法会抛出 redis.exceptions.RedisError 异常，可以在 except 块中进行处理，例如进行回滚操作（这里简单地使用 pipe.reset() 重置管道）。

9.2 乐观锁与事务

Redis 中的事务不支持传统数据库中的悲观锁机制，但可以通过 WATCH 命令实现乐观锁。WATCH 命令可以监控一个或多个键，当执行 EXEC 命令时，如果被监控的键在事务开始后被其他客户端修改，事务将被取消并返回错误。

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 监控键 'counter'
r.watch('counter')

try:
    # 获取当前值
    current_value = r.get('counter')
    if current_value is None:
        current_value = 0
    else:
        current_value = int(current_value.decode('utf - 8'))

    # 创建管道对象
    pipe = r.pipeline()

    # 开始事务
    pipe.multi()

    # 执行操作
    new_value = current_value + 1
    pipe.set('counter', new_value)

    # 执行事务
    pipe.execute()
    print(f'成功更新 counter 为 {new_value}')
except redis.exceptions.WatchError:
    print('事务执行失败，因为 counter 被其他客户端修改')
finally:
    # 取消监控
    r.unwatch()

在上述代码中：

• r.watch('counter') 监控键 counter。
• 在获取当前值并进行计算后，开始事务并设置新值。
• 如果在事务执行前，counter 键被其他客户端修改，pipe.execute() 会抛出 redis.exceptions.WatchError 异常，捕获该异常并提示事务执行失败。
• 最后使用 r.unwatch() 取消对键的监控。

10. 发布与订阅

Redis 提供了发布与订阅（Publish/Subscribe）功能，允许客户端发布消息到指定的频道，其他订阅了该频道的客户端会收到这些消息。在 redis - py 中，可以使用 pubsub 方法来实现发布与订阅功能。

10.1 订阅频道

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 创建发布订阅对象
pubsub = r.pubsub()

# 订阅频道 'news'
pubsub.subscribe('news')

# 循环监听消息
for message in pubsub.listen():
    if message['type'] =='message':
        print(f'收到消息: {message["data"].decode("utf - 8")}')

在上述代码中：

• pubsub = r.pubsub() 创建了一个发布订阅对象。
• pubsub.subscribe('news') 订阅了频道 news。
• for message in pubsub.listen() 进入一个循环，持续监听该频道上的消息。当收到消息时，如果消息类型为 message，则打印消息内容。

10.2 发布消息

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

# 发布消息到频道 'news'
r.publish('news', '重要新闻：明天有大事件发生！')

在上述代码中，r.publish('news', '重要新闻：明天有大事件发生！') 将消息 重要新闻：明天有大事件发生！ 发布到频道 news 上，订阅了该频道的客户端会收到此消息。

通过以上内容，你已经对使用 Python 操作 Redis 数据库有了较为全面的了解。在实际应用中，可以根据具体的业务需求，灵活运用 Redis 的各种数据结构和功能，充分发挥 Redis 的高性能和灵活性，提升应用程序的性能和扩展性。同时，还需要注意 Redis 服务器的配置、性能优化以及数据持久化等方面的问题，以确保 Redis 数据库的稳定运行和数据安全。