Redis 作为一款高性能的内存数据库，在缓存、会话存储、实时分析等场景中应用广泛。本文将详细介绍如何在 Python 中操作 Redis，涵盖环境搭建、基础数据类型操作、高级功能及最佳实践，适合 Redis 初学者和 Python 开发者参考。
一、环境准备
1. 安装 Redis 服务器
- Windows：从Redis 官方仓库下载安装包，或使用 WSL2 安装 Linux 版本
- Linux：通过包管理器安装sudo apt install redis-server（Ubuntu）或yum install redis（CentOS）
- 验证安装：启动服务后执行redis-cli ping，返回PONG表示运行正常
2. 安装 Python 客户端
使用官方推荐的redis-py库：
pip install redis
# 如需操作Redis集群，额外安装
pip install redis-py-cluster
二、建立连接
1. 基础连接方式
import redis
# 基本连接配置
r = redis.Redis(
host='localhost', # 服务器地址，默认localhost
port=6379, # 端口号，默认6379
password='your_pwd', # 认证密码，无密码可省略
db=0, # 数据库编号（0-15），默认0
decode_responses=True # 自动将返回值转为字符串（默认返回bytes）
)
# 测试连接
try:
if r.ping():
print("✅ Redis连接成功")
except redis.ConnectionError:
print("❌ Redis连接失败")
2. 连接池（推荐）
连接池能复用连接，减少频繁创建连接的开销：
# 创建连接池
pool = redis.ConnectionPool(
host='localhost',
port=6379,
max_connections=20, # 最大连接数
decode_responses=True
)
# 从连接池获取连接
r = redis.Redis(connection_pool=pool)
三、核心数据类型操作
1. String（字符串）
字符串是 Redis 最基础的数据类型，可存储文本、数字等，最大容量 512MB。
# 设置值
r.set('username', 'zhangsan') # 基本设置
r.setex('verify_code', 300, '654321') # 设置5分钟过期的验证码
r.mset({'age': '25', 'gender': 'male'}) # 批量设置
# 获取值
print(r.get('username')) # 输出: zhangsan
print(r.mget(['age', 'gender'])) # 批量获取: ['25', 'male']
# 数值操作
r.set('views', 100)
r.incr('views') # 自增1 → 101
r.incrby('views', 5) # 增加5 → 106
r.decr('views', 3) # 减少3 → 103
2. Hash（哈希）
哈希适合存储对象型数据（如用户信息），由字段 - 值对组成，类似 Python 字典。
# 设置字段
r.hset('user:1001', 'name', '李四') # 单个字段设置
r.hset('user:1001', mapping={ # 批量设置
'age': 30,
'city': '深圳',
'email': 'lisi@example.com'
})
# 获取数据
print(r.hget('user:1001', 'name')) # 获取单个字段 → 李四
print(r.hgetall('user:1001')) # 获取所有字段 → 完整字典
print(r.hkeys('user:1001')) # 获取所有字段名 → ['name','age',...]
print(r.hvals('user:1001')) # 获取所有值 → ['李四','30',...]
# 其他操作
r.hdel('user:1001', 'email') # 删除字段
print(r.hexists('user:1001', 'city')) # 判断字段是否存在 → True
3. List（列表）
Redis 列表是有序可重复的元素集合，底层实现为双向链表，适合实现队列、栈等结构。
# 添加元素
r.lpush('fruit', 'apple', 'banana') # 左侧插入 → [banana, apple]
r.rpush('fruit', 'orange') # 右侧插入 → [banana, apple, orange]
# 获取元素
print(r.lrange('fruit', 0, -1)) # 获取所有元素 → 完整列表
print(r.lindex('fruit', 1)) # 获取索引1的元素 → apple
print(r.llen('fruit')) # 获取长度 → 3
# 移除元素
print(r.lpop('fruit')) # 左侧弹出 → banana
print(r.rpop('fruit')) # 右侧弹出 → orange
4. Set（集合）
无序且唯一的元素集合，支持交集、并集等集合运算，适合标签、好友关系等场景。
# 添加元素
r.sadd('tags', 'python', 'redis', 'database') # 添加多个元素
# 获取数据
print(r.smembers('tags')) # 查看所有元素 → 无序集合
print(r.scard('tags')) # 元素数量 → 3
print(r.sismember('tags', 'java')) # 判断是否存在 → False
# 集合运算
r.sadd('tags2', 'redis', 'mysql', 'python')
print(r.sinter('tags', 'tags2')) # 交集 → {'python', 'redis'}
print(r.sunion('tags', 'tags2')) # 并集 → {'python','redis','database','mysql'}
5. Sorted Set（有序集合）
在集合基础上为每个元素分配分数（score），支持按分数排序和范围查询，适合排行榜、计分系统。
# 添加元素（分数, 成员）
r.zadd('ranking', {
'张三': 95,
'李四': 88,
'王五': 92
})
# 范围查询
# 按分数升序取前2名
print(r.zrange('ranking', 0, 1, withscores=True)) # [(李四,88.0), (王五,92.0)]
# 按分数降序取所有
print(r.zrevrange('ranking', 0, -1, withscores=True)) # [(张三,95.0), (王五,92.0), (李四,88.0)]
# 分数操作
print(r.zscore('ranking', '张三')) # 获取分数 → 95.0
r.zincrby('ranking', 3, '李四') # 增加分数 → 91.0
四、高级功能
1. 事务处理
使用pipeline实现事务，确保多个命令的原子性执行：
try:
with r.pipeline() as pipe:
# 开启事务
pipe.multi()
# 批量添加命令
pipe.set('key1', 'val1')
pipe.hset('user:1002', 'name', 'wangwu')
pipe.incr('counter')
# 执行事务
results = pipe.execute()
print(f"事务执行成功: {results}")
except redis.RedisError as e:
print(f"事务执行失败: {e}")
2. 发布订阅机制
实现简单的消息队列功能：
# 发布者
def publisher():
r.publish('news', 'Redis发布订阅示例')
r.publish('news', 'Python操作Redis指南')
# 订阅者
def subscriber():
pubsub = r.pubsub()
pubsub.subscribe('news') # 订阅频道
for message in pubsub.listen():
if message['type'] == 'message':
print(f"收到消息: {message['data']}")
# 可添加退出条件
if message['data'] == 'exit':
break
# 实际使用时需多线程运行
import threading
threading.Thread(target=subscriber, daemon=True).start()
publisher()
3. 过期时间管理
# 设置过期时间
r.set('temp_data', '临时数据')
r.expire('temp_data', 60) # 60秒后过期
# 查看剩余时间（-1：永不过期，-2：已过期）
print(r.ttl('temp_data')) # 剩余秒数
# 清除过期时间
r.persist('temp_data')
五、异常处理与最佳实践
1. 常见异常处理
try:
# 执行Redis操作
r.set('key', 'value')
except redis.ConnectionError:
print("连接异常：请检查Redis服务是否运行")
except redis.AuthenticationError:
print("认证失败：密码错误")
except redis.ResponseError as e:
print(f"命令执行错误：{e}")
except Exception as e:
print(f"其他错误：{e}")
2. 性能优化建议
- 使用连接池：减少 TCP 连接开销，推荐设置max_connections=10-50
- 批量操作：优先使用mset、mget等批量命令，减少网络往返
- 合理设置数据结构：例如存储用户信息用 Hash 而非多个 String
- 避免大键：单个键值不宜超过 100MB，大列表建议分片存储
- 定期清理过期数据：利用EXPIRE自动淘汰无用数据
六、总结
本文介绍了 Python 操作 Redis 的核心方法，从基础连接到高级功能，覆盖了实际开发中的常见场景。Redis 作为高性能的内存数据库，结合 Python 的简洁语法，能极大提升数据处理效率。建议在实际项目中根据业务场景选择合适的数据结构，并遵循最佳实践确保系统稳定性。
后续可深入学习 Redis 的分布式锁、Lua 脚本、集群部署等高级主题，进一步发挥 Redis 的强大功能.