Python 操作 Redis 核心指南与代码实战
环境准备与连接管理
基础依赖安装
在 Python 生态中，与 Redis 交互的标准库是 redis-py。可以通过包管理工具进行安装：
pip install redis客户端连接策略
redis-py 提供了 Redis 和 StrictRedis 两个核心类。在现代开发中，通常直接使用 Redis 类，它已经包含了所有官方命令的支持并保持了向后兼容性。
1. 基础直连模式
适用于简单的脚本或单次任务，每次操作都会建立新的 TCP 连接。
import redis
# 实例化客户端，默认连接本地 127.0.0.1:6379
db_client = redis.Redis(decode_responses=True)
# 若需连接远程节点，可指定 host 和 port
# db_client = redis.Redis(host='192.168.1.50', port=6379, db=0)
db_client.set('system:status', 'running')
print(db_client.get('system:status'))2. 连接池模式（推荐）
在生产环境中，频繁创建和销毁连接会带来显著的性能损耗。使用连接池（Connection Pool）可以复用 TCP 连接，大幅提升吞吐量。
import redis
# 初始化连接池，配置最大连接数和超时时间
pool = redis.ConnectionPool(
host='192.168.1.50',
port=6379,
max_connections=50,
decode_responses=True
)
# 将连接池注入到 Redis 实例中
cache_client = redis.Redis(connection_pool=pool)
cache_client.set('app:version', '2.4.1')
print(cache_client.get('app:version'))Django 框架集成
对于 Django 项目，推荐使用 django-redis 组件，它能无缝接管 Django 的缓存后端。
安装依赖：
pip install django-redis在 settings.py 中配置缓存后端：
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://192.168.1.50:6379/1",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {
"max_connections": 100,
"retry_on_timeout": True
}
}
}
}在业务代码中获取原生 Redis 连接：
from django_redis import get_redis_connection
# 获取配置中 default 对应的原生 Redis 客户端
raw_client = get_redis_connection("default")
raw_client.set("feature:flag:new_ui", "true", ex=3600)
raw_client.set("feature:flag:legacy", "false", timeout=None) # 永不过期基础数据结构操作
字符串 (String)
String 是 Redis 最基础的数据类型，本质上是二进制安全的字节序列，常用于缓存、计数器和分布式锁。
# 设置单个键值对，支持过期时间和条件设置
# ex: 秒级过期, px: 毫秒级过期, nx: 仅当键不存在时设置, xx: 仅当键存在时设置
cache_client.set('user:1001:token', 'abc123xyz', ex=7200, nx=True)
# 批量设置多个键值对
cache_client.mset({
'config:theme': 'dark',
'config:lang': 'zh-CN'
})
# 获取单个值
token = cache_client.get('user:1001:token')
# 批量获取多个值
configs = cache_client.mget(['config:theme', 'config:lang'])
# 获取旧值并同时设置新值（原子操作）
old_token = cache_client.getset('user:1001:token', 'new_token_456')
# 获取字符串的局部切片（基于字节索引）
# 假设值为 "HelloWorld"，获取索引 0 到 4 的字节
sub_str = cache_client.getrange('user:1001:token', 0, 4)哈希表 (Hash)
Hash 结构非常适合存储对象属性，它将一个键映射到多个字段-值对上，避免了将整个对象序列化为 JSON 字符串的开销。
# 设置单个字段
cache_client.hset('user:1001:profile', 'nickname', 'DevMaster')
# 批量设置多个字段（现代 redis-py 推荐直接使用 hset 传入 mapping）
cache_client.hset('user:1001:profile', mapping={
'age': '28',
'role': 'admin',
'status': 'active'
})
# 仅当字段不存在时设置（hsetnx）
cache_client.hsetnx('user:1001:profile', 'nickname', 'Guest') # 不会覆盖
# 获取单个字段值
nickname = cache_client.hget('user:1001:profile', 'nickname')
# 批量获取多个字段值
fields_data = cache_client.hmget('user:1001:profile', ['age', 'role'])
# 获取所有字段和值
all_profile = cache_client.hgetall('user:1001:profile')
# 获取字段数量、所有字段名、所有字段值
field_count = cache_client.hlen('user:1001:profile')
all_keys = cache_client.hkeys('user:1001:profile')
all_vals = cache_client.hvals('user:1001:profile')
# 检查字段是否存在
is_admin = cache_client.hexists('user:1001:profile', 'role')
# 删除指定字段
cache_client.hdel('user:1001:profile', 'status')
# 数值自增/自减（支持整数和浮点数）
cache_client.hincrby('user:1001:profile', 'login_count', 1)
cache_client.hincrbyfloat('user:1001:profile', 'balance', -15.50)
# 游标迭代获取（防止大 Hash 导致内存阻塞）
cursor = 0
while True:
cursor, data = cache_client.hscan('user:1001:profile', cursor=cursor, count=50)
print(data)
if cursor == 0:
break
# 使用生成器进行更优雅的迭代
for field, value in cache_client.hscan_iter('user:1001:profile', match='*count*'):
print(f"{field}: {value}")列表 (List)
List 是一个双向链表结构，支持从两端推入和弹出元素，常用于实现消息队列、时间线或最新列表。
# 从左侧（头部）推入元素
cache_client.lpush('task:queue', 'task_A', 'task_B', 'task_C')
# 当前顺序: task_C, task_B, task_A
# 从右侧（尾部）推入元素
cache_client.rpush('task:queue', 'task_D')
# 在指定元素前/后插入新元素
cache_client.linsert('task:queue', 'BEFORE', 'task_B', 'task_A.5')
# 根据索引修改元素值
cache_client.lset('task:queue', 0, 'task_C_updated')
# 删除指定元素 (num=0 删除所有, num>0 从左向右删, num<0 从右向左删)
cache_client.lrem('task:queue', 1, 'task_A')
# 获取指定索引范围的元素
tasks = cache_client.lrange('task:queue', 0, -1) # 获取全部
# 修剪列表，只保留指定索引范围内的元素
cache_client.ltrim('task:queue', 0, 99) # 只保留前 100 个元素
# 从左侧弹出元素（阻塞/非阻塞）
first_task = cache_client.lpop('task:queue')
# 获取指定索引的元素及列表总长度
second_task = cache_client.lindex('task:queue', 1)
total_tasks = cache_client.llen('task:queue')
# 自定义分块迭代器（优化大列表的遍历性能）
def chunked_list_iter(client, list_key, chunk_size=100):
"""通过 lrange 分块读取，避免 O(N^2) 的 lindex 遍历"""
total_len = client.llen(list_key)
for start in range(0, total_len, chunk_size):
end = start + chunk_size - 1
yield from client.lrange(list_key, start, end)
for item in chunked_list_iter(cache_client, 'task:queue'):
print(item)高级数据结构操作
无序集合 (Set)
Set 是一个无序且元素唯一的集合，底层基于哈希表实现。常用于去重、标签系统以及集合的交并差运算。
# 添加元素
cache_client.sadd('user:1001:tags', 'python', 'redis', 'backend')
# 获取集合基数（元素个数）
tag_count = cache_client.scard('user:1001:tags')
# 集合运算：差集、交集、并集
cache_client.sadd('user:1002:tags', 'redis', 'golang', 'devops')
diff_tags = cache_client.sdiff('user:1001:tags', 'user:1002:tags') # 1001独有
inter_tags = cache_client.sinter('user:1001:tags', 'user:1002:tags') # 共同拥有
union_tags = cache_client.sunion('user:1001:tags', 'user:1002:tags') # 所有标签
# 将运算结果直接存储到新集合
cache_client.sdiffstore('diff:result', 'user:1001:tags', 'user:1002:tags')
# 检查元素是否存在
has_python = cache_client.sismember('user:1001:tags', 'python')
# 获取所有成员
all_tags = cache_client.smembers('user:1001:tags')
# 移动元素到另一个集合
cache_client.smove('user:1001:tags', 'archive:tags', 'backend')
# 随机弹出或获取元素
popped_tag = cache_client.spop('user:1001:tags')
random_tags = cache_client.srandmember('user:1001:tags', 2) # 随机获取2个但不删除
# 删除指定元素
cache_client.srem('user:1001:tags', 'redis')
# 增量迭代
for tag in cache_client.sscan_iter('user:1001:tags', match='*'):
print(tag)有序集合 (Sorted Set)
Sorted Set 在 Set 的基础上为每个元素关联了一个分数（Score），元素按分数从小到大自动排序。适用于排行榜、延迟队列等场景。
# 添加元素及分数
cache_client.zadd('leaderboard:game1', {'player_A': 150, 'player_B': 320, 'player_C': 210})
# 获取元素总数
total_players = cache_client.zcard('leaderboard:game1')
# 统计指定分数区间的元素数量
active_players = cache_client.zcount('leaderboard:game1', 200, 500)
# 增加指定元素的分数
cache_client.zincrby('leaderboard:game1', 50, 'player_A')
# 获取指定元素的分数
score_b = cache_client.zscore('leaderboard:game1', 'player_B')
# 按索引范围获取元素（默认升序）
# withscores=True 表示同时返回分数
top_players = cache_client.zrange('leaderboard:game1', 0, 2, desc=True, withscores=True)
# 按分数范围获取元素
mid_tier = cache_client.zrangebyscore('leaderboard:game1', 100, 300)
# 获取元素的排名（从0开始，desc=True 表示按分数从高到低排）
rank_a = cache_client.zrevrank('leaderboard:game1', 'player_A')
# 字典序范围查询（当所有元素分数相同时生效）
cache_client.zadd('dict:set', {'apple': 0, 'banana': 0, 'cherry': 0, 'date': 0})
# 获取 'b' 到 'd' 之间的元素（闭区间）
lex_items = cache_client.zrangebylex('dict:set', b'[b', b'[d')
# 删除指定元素
cache_client.zrem('leaderboard:game1', 'player_C')
# 按排名范围删除（例如清理排行榜末尾）
cache_client.zremrangebyrank('leaderboard:game1', 0, 9) # 删除前10名
# 按分数范围删除
cache_client.zremrangebyscore('leaderboard:game1', 0, 100)
# 有序集合交集运算（分数聚合方式可选 SUM, MIN, MAX）
cache_client.zinterstore('result:inter', ['leaderboard:game1', 'leaderboard:game2'], aggregate='MAX')
# 增量迭代获取
for member, score in cache_client.zscan_iter('leaderboard:game1'):
print(f"{member}: {score}")全局与通用命令
除了针对特定数据结构的操作，Redis 还提供了一系列用于键空间管理的全局命令。
# 删除一个或多个键
cache_client.delete('user:1001:token', 'config:theme')
# 检查键是否存在
is_exist = cache_client.exists('user:1001:profile')
# 模式匹配查找键（生产环境慎用，可能阻塞主线程）
matching_keys = cache_client.keys('user:1001:*')
# 设置键的过期时间（秒）
cache_client.expire('user:1001:profile', 86400)
# 键重命名
cache_client.rename('old:key:name', 'new:key:name')
# 将键移动到另一个数据库
cache_client.move('system:status', 2)
# 随机返回一个键（不删除）
random_key = cache_client.randomkey()
# 获取键对应值的数据类型 (string, list, set, zset, hash 等)
key_type = cache_client.type('user:1001:profile')
# 游标迭代扫描键空间（替代 keys 命令的安全做法）
cursor = 0
while True:
cursor, keys = cache_client.scan(cursor=cursor, match='user:*', count=100)
for k in keys:
print(k)
if cursor == 0:
break
# 使用生成器扫描
for key in cache_client.scan_iter(match='config:*'):
print(key)