Python Redis 全面教程：常用 API 串联与实战指南
·
大家好，我是jobleap.cn的小九。
Redis 是一款高性能的键值对数据库，支持多种数据结构，广泛应用于缓存、消息队列、排行榜等场景。Python 的 redis 库是官方推荐的客户端，本文将从环境准备、连接方式、核心数据类型 API、通用操作、事务/管道、发布订阅等维度，全面讲解其常用用法，并通过实战案例串联所有核心 API。
一、环境准备
1. 安装 redis 库
使用 pip 安装 Python 的 redis 客户端：
# 安装稳定版
pip install redis
# 若需支持 Redis 6+ 的新特性（如 ACL），可安装最新版
pip install redis==5.0.1 # 推荐版本
2. 启动 Redis 服务
确保本地或远程 Redis 服务已启动（本地测试推荐使用 Docker 快速启动）：
# Docker 启动 Redis（无需手动配置）
docker run -d -p 6379:6379 --name redis-test redis:7.0
二、Redis 连接方式
redis 库提供两种核心连接方式：普通连接 和 连接池（推荐生产环境使用）。
1. 普通连接（基础用法）
适用于简单场景，每次操作创建/关闭连接（效率较低）：
import redis
# 1. 创建普通连接
r = redis.Redis(
host="localhost", # Redis 地址
port=6379, # 端口
db=0, # 数据库编号（Redis 默认 16 个库，0-15）
password="", # 若设置了密码需填写
decode_responses=True # 自动将返回值从 bytes 转为字符串（关键！）
)
# 测试连接
try:
r.ping() # 发送 ping 指令，返回 PONG 则连接成功
print("Redis 连接成功！")
except Exception as e:
print(f"连接失败：{e}")
2. 连接池（生产环境推荐）
连接池复用连接，避免频繁创建/销毁连接的开销：
import redis
# 1. 创建连接池
pool = redis.ConnectionPool(
host="localhost",
port=6379,
db=0,
decode_responses=True,
max_connections=10 # 最大连接数
)
# 2. 从连接池获取连接
r = redis.Redis(connection_pool=pool)
# 测试连接
print(r.ping()) # 输出：True
三、核心数据类型 API 详解
Redis 支持 5 种核心数据类型：String（字符串）、Hash（哈希）、List（列表）、Set（集合）、ZSet（有序集合），以下逐个讲解其常用 API。
1. String（字符串）
最基础的类型，适用于缓存、计数器、分布式锁等场景。
| API | 作用 | 示例 |
|---|---|---|
set(key, value, ex=None) |
设置键值对，ex 为过期时间（秒） | r.set("name", "张三", ex=60) |
get(key) |
获取键值 | r.get("name") → “张三” |
mset(**kwargs) |
批量设置键值对 | r.mset(age=20, city="北京") |
mget(*keys) |
批量获取键值 | r.mget("name", "age") → [“张三”, “20”] |
incr(key, amount=1) |
整数自增（amount 为步长） | r.incr("count") → 1；r.incr("count", 2) → 3 |
decr(key, amount=1) |
整数自减 | r.decr("count") → 2 |
append(key, value) |
追加字符串 | r.append("name", "丰") → “张三丰” |
setex(key, time, value) |
设置带过期时间的键值（等价于 set + expire） | r.setex("code", 120, "123456") |
完整示例：
# 1. 单个设置/获取
r.set("name", "张三")
print(r.get("name")) # 输出：张三
# 2. 批量设置/获取
r.mset(age=20, city="北京")
print(r.mget("name", "age", "city")) # 输出：['张三', '20', '北京']
# 3. 计数器
r.set("count", 0)
r.incr("count") # 自增 1 → 1
r.incr("count", 5) # 自增 5 → 6
r.decr("count") # 自减 1 → 5
print(r.get("count")) # 输出：5
# 4. 过期设置
r.setex("verify_code", 120, "8888") # 2 分钟过期
print(r.ttl("verify_code")) # 查看剩余过期时间（秒）→ 约 120
2. Hash（哈希）
键值对的集合，适用于缓存对象（如用户信息、商品信息）。
| API | 作用 | 示例 |
|---|---|---|
hset(name, key, value) |
设置哈希字段值 | r.hset("user:1", "name", "李四") |
hget(name, key) |
获取哈希字段值 | r.hget("user:1", "name") → “李四” |
hmset(name, mapping) |
批量设置哈希字段（注意：redis 4.0+ 推荐用 hset） | r.hset("user:1", mapping={"age":25, "city":"上海"}) |
hmget(name, *keys) |
批量获取哈希字段 | r.hmget("user:1", "name", "age") → [“李四”, “25”] |
hgetall(name) |
获取哈希所有字段和值 | r.hgetall("user:1") → {“name”:“李四”, “age”:“25”, …} |
hkeys(name) |
获取哈希所有字段名 | r.hkeys("user:1") → [“name”, “age”, “city”] |
hvals(name) |
获取哈希所有值 | r.hvals("user:1") → [“李四”, “25”, “上海”] |
hdel(name, *keys) |
删除哈希指定字段 | r.hdel("user:1", "city") |
hincrby(name, key, amount=1) |
哈希字段值自增 | r.hincrby("user:1", "age", 1) → 26 |
完整示例：
# 1. 设置用户信息（哈希）
r.hset("user:1", "name", "李四")
r.hset("user:1", mapping={"age":25, "city":"上海"})
# 2. 获取单个/多个字段
print(r.hget("user:1", "name")) # 输出：李四
print(r.hmget("user:1", "name", "age")) # 输出：['李四', '25']
# 3. 获取所有字段和值
print(r.hgetall("user:1")) # 输出：{'name': '李四', 'age': '25', 'city': '上海'}
# 4. 字段自增
r.hincrby("user:1", "age", 1)
print(r.hget("user:1", "age")) # 输出：26
# 5. 删除字段
r.hdel("user:1", "city")
print(r.hgetall("user:1")) # 输出：{'name': '李四', 'age': '26'}
3. List（列表）
有序、可重复的字符串列表，适用于消息队列、最新消息列表等场景。
| API | 作用 | 示例 |
|---|---|---|
lpush(name, *values) |
从列表左侧插入元素 | r.lpush("msg_list", "消息1", "消息2") |
rpush(name, *values) |
从列表右侧插入元素 | r.rpush("msg_list", "消息3") |
lpop(name) |
从列表左侧弹出元素 | r.lpop("msg_list") → “消息2” |
rpop(name) |
从列表右侧弹出元素 | r.rpop("msg_list") → “消息3” |
lrange(name, start, end) |
获取列表指定范围元素（end=-1 表示最后一个） | r.lrange("msg_list", 0, -1) → [“消息1”] |
llen(name) |
获取列表长度 | r.llen("msg_list") → 1 |
lindex(name, index) |
获取列表指定索引元素 | r.lindex("msg_list", 0) → “消息1” |
lrem(name, count, value) |
删除列表中指定元素（count=0 删所有，>0 从左删，<0 从右删） | r.lrem("msg_list", 0, "消息1") |
完整示例：
# 1. 插入元素（左/右）
r.lpush("msg_list", "消息1", "消息2") # 列表：[消息2, 消息1]
r.rpush("msg_list", "消息3") # 列表：[消息2, 消息1, 消息3]
# 2. 获取列表长度和范围
print(r.llen("msg_list")) # 输出：3
print(r.lrange("msg_list", 0, -1)) # 输出：['消息2', '消息1', '消息3']
# 3. 弹出元素
print(r.lpop("msg_list")) # 输出：消息2（左侧弹出）
print(r.rpop("msg_list")) # 输出：消息3（右侧弹出）
print(r.lrange("msg_list", 0, -1)) # 输出：['消息1']
# 4. 删除指定元素
r.lrem("msg_list", 0, "消息1")
print(r.llen("msg_list")) # 输出：0
4. Set（集合）
无序、不可重复的字符串集合，适用于去重、交集/并集/差集计算（如共同好友）。
| API | 作用 | 示例 |
|---|---|---|
sadd(name, *values) |
向集合添加元素 | r.sadd("user:1:friends", "2", "3", "4") |
smembers(name) |
获取集合所有元素 | r.smembers("user:1:friends") → {“2”,“3”,“4”} |
sismember(name, value) |
判断元素是否在集合中 | r.sismember("user:1:friends", "2") → True |
srem(name, *values) |
从集合删除元素 | r.srem("user:1:friends", "4") |
scard(name) |
获取集合元素个数 | r.scard("user:1:friends") → 2 |
sinter(*names) |
求多个集合的交集 | r.sinter("user:1:friends", "user:2:friends") |
sunion(*names) |
求多个集合的并集 | r.sunion("user:1:friends", "user:2:friends") |
sdiff(*names) |
求多个集合的差集 | r.sdiff("user:1:friends", "user:2:friends") |
完整示例：
# 1. 添加集合元素
r.sadd("user:1:friends", "2", "3", "4")
r.sadd("user:2:friends", "3", "4", "5")
# 2. 查看集合元素
print(r.smembers("user:1:friends")) # 输出：{'2', '3', '4'}
print(r.scard("user:1:friends")) # 输出：3
# 3. 判断元素是否存在
print(r.sismember("user:1:friends", "2")) # 输出：True
# 4. 集合运算（共同好友、所有好友、独有好友）
common_friends = r.sinter("user:1:friends", "user:2:friends")
all_friends = r.sunion("user:1:friends", "user:2:friends")
unique_friends = r.sdiff("user:1:friends", "user:2:friends")
print("共同好友：", common_friends) # 输出：{'3', '4'}
print("所有好友：", all_friends) # 输出：{'2', '3', '4', '5'}
print("用户1独有好友：", unique_friends) # 输出：{'2'}
# 5. 删除元素
r.srem("user:1:friends", "2")
print(r.smembers("user:1:friends")) # 输出：{'3', '4'}
5. ZSet（有序集合）
有序、不可重复的集合，每个元素关联一个分数（score），按分数排序，适用于排行榜、延迟队列等场景。
| API | 作用 | 示例 |
|---|---|---|
zadd(name, mapping) |
添加元素（mapping 为 {元素: 分数}） | r.zadd("rank", {"张三": 90, "李四": 85}) |
zrange(name, start, end, withscores=False) |
按分数升序获取元素（withscores=True 带分数） | r.zrange("rank", 0, -1, withscores=True) |
zrevrange(name, start, end, withscores=False) |
按分数降序获取元素 | r.zrevrange("rank", 0, 0) → 第一名 |
zscore(name, value) |
获取元素的分数 | r.zscore("rank", "张三") → 90.0 |
zcard(name) |
获取元素个数 | r.zcard("rank") → 2 |
zincrby(name, amount, value) |
元素分数自增 | r.zincrby("rank", 5, "李四") → 90.0 |
zrem(name, *values) |
删除元素 | r.zrem("rank", "李四") |
完整示例：
# 1. 添加排行榜元素（姓名: 分数）
r.zadd("score_rank", {"张三": 90, "李四": 85, "王五": 95})
# 2. 按分数降序获取排行榜（带分数）
top3 = r.zrevrange("score_rank", 0, 2, withscores=True)
print("成绩排行榜 Top3：", top3)
# 输出：[('王五', 95.0), ('张三', 90.0), ('李四', 85.0)]
# 3. 获取单个元素分数
print("张三的分数：", r.zscore("score_rank", "张三")) # 输出：90.0
# 4. 分数自增（李四加5分）
r.zincrby("score_rank", 5, "李四")
print("李四加分后的分数：", r.zscore("score_rank", "李四")) # 输出：90.0
# 5. 获取排行榜长度
print("排行榜人数：", r.zcard("score_rank")) # 输出：3
# 6. 删除元素
r.zrem("score_rank", "王五")
print(r.zrevrange("score_rank", 0, -1)) # 输出：['张三', '李四']
四、通用操作 API
适用于所有数据类型的通用指令，如过期、删除、检查存在等。
| API | 作用 | 示例 |
|---|---|---|
expire(key, time) |
设置键过期时间（秒） | r.expire("name", 60) |
ttl(key) |
查看键剩余过期时间（-1 永不过期，-2 已过期） | r.ttl("name") → 约 60 |
persist(key) |
取消键过期时间 | r.persist("name") |
delete(*keys) |
删除键 | r.delete("name", "age") |
exists(*keys) |
检查键是否存在（返回存在的数量） | r.exists("name") → 1 |
keys(pattern) |
按通配符匹配键（慎用！生产环境可能阻塞） | r.keys("user:*") → [“user:1”, “user:2”] |
flushdb() |
清空当前数据库 | r.flushdb() |
flushall() |
清空所有数据库 | r.flushall() |
type(key) |
查看键的数据类型 | r.type("score_rank") → “zset” |
示例：
# 1. 设置过期时间
r.set("temp_key", "临时数据")
r.expire("temp_key", 30)
print("剩余过期时间：", r.ttl("temp_key")) # 输出：约 30
# 2. 取消过期时间
r.persist("temp_key")
print("取消过期后：", r.ttl("temp_key")) # 输出：-1
# 3. 检查键是否存在
print("temp_key 是否存在：", r.exists("temp_key")) # 输出：1
# 4. 删除键
r.delete("temp_key")
print("删除后是否存在：", r.exists("temp_key")) # 输出：0
# 5. 匹配键（通配符 * 匹配所有，? 匹配单个字符）
r.set("user:2", "李四")
r.set("user:3", "王五")
print("匹配 user:* 的键：", r.keys("user:*")) # 输出：['user:1', 'user:2', 'user:3']
# 6. 查看数据类型
print("score_rank 类型：", r.type("score_rank")) # 输出：zset
五、高级特性：事务与管道
1. 事务（Transaction）
Redis 事务保证指令的原子性（要么全部执行，要么全部不执行），通过 multi() 开启事务，exec() 执行，discard() 取消。
# 1. 开启事务
pipe = r.pipeline(transaction=True)
# 2. 批量添加事务指令（此时未执行）
pipe.set("a", 1)
pipe.incr("a")
pipe.set("b", 2)
# 3. 执行事务
result = pipe.execute()
print("事务执行结果：", result) # 输出：[True, 2, True]
# 4. 取消事务（示例）
pipe = r.pipeline(transaction=True)
pipe.set("c", 3)
pipe.discard() # 取消事务
print(r.get("c")) # 输出：None
2. 管道（Pipeline）
批量执行指令，减少网络往返次数，提升效率（即使不开启事务，管道也能批量操作）。
# 1. 创建管道（不开启事务）
pipe = r.pipeline(transaction=False)
# 2. 批量添加指令
pipe.set("x", 10)
pipe.incr("x")
pipe.hset("user:3", "name", "赵六")
pipe.hgetall("user:3")
# 3. 执行管道（返回所有指令的结果）
results = pipe.execute()
print("管道执行结果：", results)
# 输出：[True, 11, 1, {'name': '赵六'}]
六、发布订阅（Pub/Sub）
Redis 支持简单的发布订阅模式，适用于轻量级消息通知。
1. 订阅者（Subscriber）
import redis
# 创建订阅者连接
sub = redis.Redis(host="localhost", port=6379, decode_responses=True)
pubsub = sub.pubsub()
# 订阅频道 "notice"
pubsub.subscribe("notice")
# 循环监听消息
print("订阅者已启动，等待消息...")
for message in pubsub.listen():
# message 结构：{'type': 'message', 'channel': 'notice', 'data': '消息内容'}
if message["type"] == "message":
print(f"收到频道 {message['channel']} 的消息：{message['data']}")
# 可选：按条件退出
if message["data"] == "exit":
print("订阅者退出")
break
2. 发布者（Publisher）
import redis
# 创建发布者连接
pub = redis.Redis(host="localhost", port=6379, decode_responses=True)
# 发布消息到频道 "notice"
pub.publish("notice", "新用户注册成功！")
pub.publish("notice", "系统将在10分钟后维护")
pub.publish("notice", "exit") # 发送退出指令
print("消息发布完成")
七、实战案例：电商商品缓存系统
串联以上所有核心 API，实现一个简单的电商商品缓存系统，包含以下功能：
- 缓存商品信息（Hash）；
- 商品浏览量计数器（String）；
- 热门商品排行榜（ZSet）；
- 商品评论列表（List）；
- 商品标签去重（Set）；
- 缓存过期自动清理（expire）。
import redis
# 1. 初始化连接池和连接
pool = redis.ConnectionPool(
host="localhost",
port=6379,
db=0,
decode_responses=True
)
r = redis.Redis(connection_pool=pool)
# 2. 定义商品 ID
goods_id = 1001
# ---------------------- 功能 1：缓存商品信息（Hash） ----------------------
r.hset(
f"goods:{goods_id}",
mapping={
"name": "Python 实战教程",
"price": 59.9,
"stock": 1000,
"category": "图书"
}
)
# 设置商品信息 24 小时过期
r.expire(f"goods:{goods_id}", 86400)
print("商品信息：", r.hgetall(f"goods:{goods_id}"))
# ---------------------- 功能 2：商品浏览量计数器（String） ----------------------
r.incr(f"goods:{goods_id}:view_count") # 浏览量 +1
r.incr(f"goods:{goods_id}:view_count", 5) # 再 +5
print("商品浏览量：", r.get(f"goods:{goods_id}:view_count"))
# ---------------------- 功能 3：热门商品排行榜（ZSet） ----------------------
# 按浏览量排序
r.zadd("hot_goods_rank", {goods_id: r.get(f"goods:{goods_id}:view_count")})
# 获取热门商品 Top5
hot_top5 = r.zrevrange("hot_goods_rank", 0, 4, withscores=True)
print("热门商品 Top5：", hot_top5)
# ---------------------- 功能 4：商品评论列表（List） ----------------------
# 添加评论（右侧插入）
r.rpush(f"goods:{goods_id}:comments", "内容很实用！", "性价比高", "推荐购买")
# 获取最新 2 条评论
latest_comments = r.lrange(f"goods:{goods_id}:comments", -2, -1)
print("最新 2 条评论：", latest_comments)
# ---------------------- 功能 5：商品标签去重（Set） ----------------------
# 添加标签（自动去重）
r.sadd(f"goods:{goods_id}:tags", "Python", "编程", "教程", "Python")
print("商品标签（去重）：", r.smembers(f"goods:{goods_id}:tags"))
# ---------------------- 通用操作：删除过期测试 ----------------------
print("商品信息剩余过期时间：", r.ttl(f"goods:{goods_id}"))
八、注意事项
- decode_responses=True：务必设置，否则返回值为 bytes 类型，需手动解码；
- 连接池复用：生产环境必须使用连接池，避免频繁创建连接；
- keys 指令慎用：
keys *会遍历所有键，Redis 单线程模型下会阻塞其他操作，生产环境推荐用scan指令； - 事务不保证回滚：Redis 事务仅保证原子性，若指令语法错误，已执行的指令不会回滚；
- 过期时间单位：
expire单位为秒，pexpire单位为毫秒。
总结
本文覆盖了 Python redis 库的核心用法：连接方式、5 种核心数据类型 API、通用操作、事务/管道、发布订阅，并通过实战案例串联所有知识点。掌握这些用法后，可应对绝大多数 Redis 应用场景（缓存、计数器、排行榜、消息队列等）。如需更高级的特性（如集群、哨兵、Lua 脚本），可参考 redis 库官方文档：https://redis-py.readthedocs.io/。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
21
0- 0
已为社区贡献20条内容
所有评论(0)