前言
在 Python 后端开发中，Redis 基本属于“高频刚需型中间件”。只要系统里出现缓存、登录态、验证码、排行榜、消息通知、限流、分布式锁、异步任务队列等场景，Redis 往往都会出场。
Redis 官方文档把 Redis 定位为一种数据结构服务器，提供 String、Hash、List、Set、Sorted Set 等多种原生数据类型，可用于缓存、队列、事件处理等场景。Python 里常用的 Redis 客户端是 redis-py，官方文档推荐使用它连接 Redis，并且同时支持同步和异步 API。
一、Redis 在 Python 项目中的常见定位
Redis 在 Python 项目里通常不是“数据库替代品”，而是一个高性能的辅助型中间件。
常见的职责包括：
- 缓存热点数据：例如用户信息、商品详情、配置数据、接口返回结果。
- 保存临时状态：例如短信验证码、登录 token、图形验证码、一次性链接。
- 计数器：例如文章阅读量、接口访问次数、点赞数、库存扣减前的预校验。
- 排行榜
- 简单消息队列：例如用 List 实现任务队列，用 Pub/Sub 做消息通知。
- 分布式锁：例如防止重复提交、防止多个服务同时执行定时任务。
- 限流
Redis 的优点是速度快、数据结构丰富、操作简单；缺点是不能无脑当主数据库使用，尤其是涉及强一致性、复杂事务、复杂查询时，还是应该交给 MySQL、PostgreSQL 等关系型数据库。
二、环境准备
1. 安装 Python Redis 客户端
官方推荐安装方式：
如果希望获得更好的解析性能，可以安装 hiredis 支持：
pip install "redis[hiredis]"
如果安装了 hiredis >= 1.0，redis-py 会尝试使用它作为响应解析器，通常不需要修改业务代码。
2. 启动 Redis
本地使用 Docker 启动 Redis：
dockerrun -d \ --name redis-demo \ -p 6379:6379 \ redis:7
查看容器：
输出示例：
CONTAINERIDIMAGECOMMANDCREATEDSTATUSPORTSNAMESf4b19e8a23c1redis:7"docker-entrypoint.s…"12secondsagoUp11seconds0.0.0.0:6379->6379/tcpredis-demo
测试连接：
docker exec -it redis-demo redis-cli ping
输出：
三、Python 连接 Redis
1. 最简单连接方式
importredisr = redis.Redis(host="localhost",port=6379,db=0,decode_responses=True)result = r.ping()print(result)
输出：
重要参数：decode_responses=True
如果不加该参数，Redis 返回的字符串通常是 bytes 类型。
import redisr = redis.Redis(host="localhost", port=6379, db=0)r.set("name", "vito")print(r.get("name"))
输出：
加上 decode_responses=True 后：
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)r.set("name", "vito")print(r.get("name"))
输出：
在大多数业务代码中，建议加上 decode_responses=True，否则后续处理字符串时经常会遇到 bytes 和 str 混用的问题。
四、String 操作：缓存、计数器、验证码
String 是 Redis 最常用的数据类型，可以存普通字符串、JSON 字符串、数字等。
1. set / get
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)r.set("user:1001:name", "张三")name = r.get("user:1001:name")print("用户名称:", name)
输出：
2. 设置过期时间
常见场景：短信验证码 5 分钟过期。
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)r.set("sms:code:13800138000", "9527", ex=300)code = r.get("sms:code:13800138000")ttl = r.ttl("sms:code:13800138000")print("验证码:", code)print("剩余过期时间:", ttl)
输出示例：
过几秒后再次执行：
3. 自增计数器
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)r.delete("article:1001:views")print(r.incr("article:1001:views"))print(r.incr("article:1001:views"))print(r.incr("article:1001:views"))
输出：
非常适合做访问量统计。
4. 接口访问次数统计
importredisr = redis.Redis(host="localhost",port=6379,db=0,decode_responses=True)user_id = 1001key = f"api:visit:{user_id}"count = r.incr(key)ifcount == 1:r.expire(key,60)print(f"用户{user_id} 当前 60 秒内访问次数: {count}")
连续运行几次：
用户1001当前60秒内访问次数:1用户1001当前60秒内访问次数:2用户1001当前60秒内访问次数:3
这就是一个最基础的限流雏形。
五、Hash 操作：保存对象数据
Hash 很适合保存一个对象的多个字段，例如用户信息、商品信息、订单摘要。
1. hset / hget / hgetall
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)user_key = "user:1001"r.hset(user_key, mapping={"name": "张三","age": "28","city": "深圳","role": "tester"})print("姓名:", r.hget(user_key, "name"))print("年龄:", r.hget(user_key, "age"))print("完整用户信息:", r.hgetall(user_key))
输出：
姓名: 张三年龄: 28完整用户信息: {'name': '张三', 'age': '28', 'city': '深圳', 'role': 'tester'}
2. 修改单个字段
r.hset("user:1001", "city", "广州")print(r.hgetall("user:1001"))
输出：
{'name': '张三', 'age': '28', 'city': '广州', 'role': 'tester'}
3. 判断字段是否存在
print(r.hexists("user:1001", "name"))print(r.hexists("user:1001", "email"))
输出：
4. 删除字段
r.hdel("user:1001", "role")print(r.hgetall("user:1001"))
输出：
{'name': '张三', 'age': '28', 'city': '广州'}
5. Hash 适合什么场景？
适合：
- user:1001 -> name, age, city
- product:2001 -> title, price, stock
- order:3001 -> status, amount, create_time
不适合：
Redis Hash 是简单对象存储，不是关系型数据库。
六、List 操作：队列、栈、任务列表
Redis List 是字符串值组成的链表，经常用于实现栈、队列和后台任务队列。
1. 使用 List 实现队列
生产任务：
import redisimport jsonr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)queue_key = "task:queue"r.delete(queue_key)tasks = [ {"task_id": 1, "type": "send_email", "email": "a@example.com"}, {"task_id": 2, "type": "send_sms", "phone": "13800138000"}, {"task_id": 3, "type": "generate_report", "report_id": 9001},]for task in tasks: r.rpush(queue_key, json.dumps(task, ensure_ascii=False)) print("任务入队:", task)
输出：
任务入队: {'task_id': 1, 'type': 'send_email', 'email': 'a@example.com'}任务入队: {'task_id': 2, 'type': 'send_sms', 'phone': '13800138000'}任务入队: {'task_id': 3, 'type': 'generate_report', 'report_id': 9001}
消费任务：
import redisimport jsonr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)queue_key = "task:queue"whileTrue: task_json = r.lpop(queue_key)if task_json isNone: print("队列为空，暂无任务")break task = json.loads(task_json) print("开始处理任务:", task)
输出：
开始处理任务: {'task_id': 1, 'type': 'send_email', 'email': 'a@example.com'}开始处理任务: {'task_id': 2, 'type': 'send_sms', 'phone': '13800138000'}开始处理任务: {'task_id': 3, 'type': 'generate_report', 'report_id': 9001}队列为空，暂无任务
这里使用 rpush 入队，lpop 出队，符合先进先出队列逻辑。
2. 查看队列长度
length = r.llen("task:queue")print("当前队列长度:", length)
输出：
3. 阻塞式消费
普通 lpop 如果队列没有数据，会马上返回 None。如果希望消费者阻塞等待，可以用 blpop。
import redisimport jsonr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)queue_key = "task:queue"print("等待任务中...")result = r.blpop(queue_key, timeout=10)if result isNone: print("10 秒内没有任务，退出")else: key, task_json = result task = json.loads(task_json) print("收到任务:", task)
如果 10 秒内没有任务：
如果期间有任务入队：
等待任务中...收到任务: {'task_id': 4, 'type': 'sync_data', 'table': 'user'}
七、Set 操作：去重、标签、集合关系
Set 是无序不重复集合，适合去重、标签、共同好友、权限集合等场景。
1. 添加元素
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)key = "article:1001:like_users"r.delete(key)r.sadd(key, "user_1", "user_2", "user_3")r.sadd(key, "user_2")print("点赞用户:", r.smembers(key))print("点赞人数:", r.scard(key))
输出：
点赞用户: {'user_2', 'user_1', 'user_3'}点赞人数: 3
虽然添加了两次 user_2，但 Set 会自动去重。
2. 判断是否存在
print(r.sismember("article:1001:like_users", "user_1"))print(r.sismember("article:1001:like_users", "user_9"))
输出：
3. 共同好友
r.delete("user:1:friends", "user:2:friends")r.sadd("user:1:friends", "张三", "李四", "王五")r.sadd("user:2:friends", "李四", "王五", "赵六")common = r.sinter("user:1:friends", "user:2:friends")print("共同好友:", common)
输出：
4. 差集
diff = r.sdiff("user:1:friends", "user:2:friends")print("用户1有但用户2没有的好友:", diff)
输出：
八、Sorted Set 操作：排行榜
Sorted Set（ZSet）是带分数的有序集合，非常适合做排行榜。
1. 添加排行榜数据
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)rank_key = "game:rank"r.delete(rank_key)r.zadd(rank_key, {"张三": 1200,"李四": 1800,"王五": 1500,"赵六": 2100,"钱七": 900,})top_users = r.zrevrange(rank_key, 0, 2, withscores=True)print("游戏积分 Top3:")for index, item in enumerate(top_users, start=1): name, score = item print(f"第 {index} 名: {name}, 分数: {int(score)}")
输出：
游戏积分Top3:第1名:赵六,分数:2100第2名:李四,分数:1800第3名:王五,分数:1500
2. 给用户增加分数
new_score = r.zincrby(rank_key, 300, "张三")print("张三最新分数:", int(new_score))
输出：
再次查看排行榜：
top_users = r.zrevrange(rank_key, 0, 4, withscores=True)for index, item in enumerate(top_users, start=1): name, score = item print(f"第 {index} 名: {name}, 分数: {int(score)}")
输出：
第1名:赵六,分数:2100第2名:李四,分数:1800第3名:张三,分数:1500第4名:王五,分数:1500第5名:钱七,分数:900
3. 查询某个用户排名
rank = r.zrevrank(rank_key, "张三")score = r.zscore(rank_key, "张三")print("张三排名:", rank + 1)print("张三分数:", int(score))
输出：
zrevrank 返回从 0 开始的排名，展示时通常要 +1。
九、Key 过期时间操作
1. expire 设置过期
r.set("login:token:abc123", "user_1001")r.expire("login:token:abc123", 3600)print("token:", r.get("login:token:abc123"))print("ttl:", r.ttl("login:token:abc123"))
输出：
token:user_1001ttl:3600
2. ttl 返回值含义
print(r.ttl("login:token:abc123"))print(r.ttl("not_exists_key"))
可能输出：
常见含义：
3. persist 移除过期时间
r.set("system:config", "v1")r.expire("system:config", 60)print("设置过期后 ttl:", r.ttl("system:config"))r.persist("system:config")print("移除过期后 ttl:", r.ttl("system:config"))
输出：
设置过期后ttl:60移除过期后ttl:-1
十、Pipeline：批量操作提升性能
如果一次要执行很多 Redis 命令，逐条发送会产生较多网络往返开销。Pipeline 可以把多个命令一起发送给服务端，减少网络和处理开销。
1. 普通循环写入
import redisimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)start = time.time()for i in range(1000): r.set(f"normal:key:{i}", i)end = time.time()print("普通写入耗时:", round(end - start, 4), "秒")
输出示例：
2. Pipeline 批量写入
import redisimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)start = time.time()pipe = r.pipeline()for i in range(1000): pipe.set(f"pipe:key:{i}", i)result = pipe.execute()end = time.time()print("Pipeline 写入结果数量:", len(result))print("Pipeline 写入耗时:", round(end - start, 4), "秒")
输出示例：
Pipeline写入结果数量: 1000Pipeline写入耗时: 0.0186 秒
Pipeline 在大量命令场景下通常明显更快。
3. Pipeline 返回值
pipe = r.pipeline()pipe.set("user:1:name", "张三")pipe.get("user:1:name")pipe.incr("counter:test")pipe.ttl("user:1:name")result = pipe.execute()print(result)
输出：
返回值按照命令加入 Pipeline 的顺序返回。
十一、事务：MULTI / EXEC 思想
redis-py 的 Pipeline 默认会把命令包在事务里执行，也可以通过参数关闭事务模式。
1. 简单事务示例
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)r.set("account:1001:balance", 1000)r.set("account:1002:balance", 500)pipe = r.pipeline(transaction=True)pipe.decrby("account:1001:balance", 100)pipe.incrby("account:1002:balance", 100)result = pipe.execute()print("事务执行结果:", result)print("账户1001余额:", r.get("account:1001:balance"))print("账户1002余额:", r.get("account:1002:balance"))
输出：
事务执行结果: [900, 600]账户1001余额: 900账户1002余额: 600
2. 注意：Redis 事务不是关系型数据库事务
Redis 的事务更多是“命令队列化 + 顺序执行”，并不等同于 MySQL 那种复杂事务。金额、订单、库存等核心数据的主流程仍应放在数据库中，Redis 可做缓存、预扣、限流或辅助判断。
十二、发布订阅 Pub/Sub
Pub/Sub 适合做轻量级实时通知，例如：
1. 订阅者
新建 subscriber.py：
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)pubsub = r.pubsub()pubsub.subscribe("order_channel")print("订阅 order_channel 成功，等待消息...")for message in pubsub.listen(): print("收到原始消息:", message)if message["type"] == "message": print("收到业务消息:", message["data"])
运行：
输出：
订阅 order_channel 成功，等待消息...收到原始消息: {'type': 'subscribe', 'pattern': None, 'channel': 'order_channel', 'data': 1}
2. 发布者
新建 publisher.py：
import redisimport jsonr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)message = {"order_id": "O202605100001","status": "paid","amount": 199.9}count = r.publish("order_channel", json.dumps(message, ensure_ascii=False))print("消息已发布")print("接收客户端数量:", count)
运行：
输出：
订阅者终端会看到：
收到原始消息: {'type': 'message', 'pattern': None, 'channel': 'order_channel', 'data': '{"order_id": "O202605100001", "status": "paid", "amount": 199.9}'}收到业务消息: {"order_id": "O202605100001", "status": "paid", "amount": 199.9}
3. Pub/Sub 的缺点
Pub/Sub 很轻量，但不是可靠消息队列。如果订阅者当时不在线，消息不会自动保存。因此适合在线通知、临时广播、实时刷新等场景，但不适合必须保证消费成功的可靠异步任务，这种场景建议使用 Kafka、RabbitMQ、RocketMQ 或 Redis Stream。
十三、分布式锁
分布式锁常见场景：
1. 使用 redis-py 自带 Lock
importredisimporttimer = redis.Redis(host="localhost",port=6379,db=0,decode_responses=True)lock = r.lock(name="lock:daily_report",timeout=10,blocking_timeout=3)try:acquired = lock.acquire()ifnot acquired:print("没有获取到锁，任务不执行")else:print("获取锁成功，开始执行任务")time.sleep(2)print("任务执行完成")finally:iflock.owned():lock.release()print("锁已释放")
输出：
2. 手写简化版锁
理解原理可用此示例：
import redisimport uuidimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)lock_key = "lock:submit:order:1001"lock_value = str(uuid.uuid4())locked = r.set(lock_key, lock_value, nx=True, ex=10)if locked:try:print("获取锁成功:", lock_value)print("开始处理订单提交...") time.sleep(2)print("订单提交处理完成") finally: current_value = r.get(lock_key)if current_value == lock_value: r.delete(lock_key)print("释放锁成功")else:print("锁已过期或被其他请求持有，不释放")else:print("请求正在处理中，请勿重复提交")
核心思想：set key value nx ex —— 不存在才设置，设置成功代表拿到锁，设置过期时间防止死锁，释放前确认 value 是自己的。生产环境建议使用成熟实现。
十四、缓存封装实战：查询用户信息
业务需求：查询用户信息，优先查 Redis，没有则查数据库，查到后写入 Redis。
1. 模拟数据库
import timeUSER_DB = {1001: {"id": 1001, "name": "张三", "city": "深圳", "role": "测试工程师"},1002: {"id": 1002, "name": "李四", "city": "广州", "role": "开发工程师"},}defquery_user_from_db(user_id: int): print(f"正在查询数据库，user_id={user_id}") time.sleep(0.2)return USER_DB.get(user_id)
2. Redis 缓存封装
import redisimport jsonimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)USER_DB = {1001: {"id": 1001, "name": "张三", "city": "深圳", "role": "测试工程师"},1002: {"id": 1002, "name": "李四", "city": "广州", "role": "开发工程师"},}defquery_user_from_db(user_id: int): print(f"正在查询数据库，user_id={user_id}") time.sleep(0.2)return USER_DB.get(user_id)defget_user(user_id: int): cache_key = f"cache:user:{user_id}" cache_value = r.get(cache_key)if cache_value: print("命中 Redis 缓存")return json.loads(cache_value) print("Redis 未命中") user = query_user_from_db(user_id)if user isNone: print("数据库也没有查到用户")returnNone r.set(cache_key, json.dumps(user, ensure_ascii=False), ex=300) print("用户信息已写入 Redis，过期时间 300 秒")return userprint("第一次查询:")print(get_user(1001))print("\n第二次查询:")print(get_user(1001))
输出：
第一次查询:Redis 未命中正在查询数据库，user_id=1001用户信息已写入 Redis，过期时间 300 秒{'id': 1001, 'name': '张三', 'city': '深圳', 'role': '测试工程师'}第二次查询:命中 Redis 缓存{'id': 1001, 'name': '张三', 'city': '深圳', 'role': '测试工程师'}
这就是最典型的缓存旁路模式（Cache Aside Pattern）。
十五、缓存穿透、击穿、雪崩
1. 缓存穿透
缓存穿透指请求的数据 Redis 和数据库都没有，每次请求都会打到数据库。解决方式：参数校验、缓存空值、布隆过滤器。
示例：缓存空值
import redisimport jsonimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)USER_DB = {1001: {"id": 1001, "name": "张三"}}defquery_user_from_db(user_id): print(f"查询数据库 user_id={user_id}") time.sleep(0.1)return USER_DB.get(user_id)defget_user(user_id): cache_key = f"cache:user:{user_id}" cache_value = r.get(cache_key)if cache_value isnotNone:if cache_value == "": print("命中空值缓存")returnNone print("命中正常缓存")return json.loads(cache_value) print("缓存未命中") user = query_user_from_db(user_id)if user isNone: r.set(cache_key, "", ex=60) print("数据库无数据，写入空值缓存 60 秒")returnNone r.set(cache_key, json.dumps(user, ensure_ascii=False), ex=300)return userprint(get_user(9999))print(get_user(9999))
输出：
缓存未命中查询数据库 user_id=9999数据库无数据，写入空值缓存 60 秒None命中空值缓存None
2. 缓存击穿
热点 key 过期，大量请求同时打向数据库。解决方式：热点 key 不设过期、互斥锁回源、提前异步刷新。
简化版互斥锁示例：
import redisimport jsonimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)defquery_hot_data_from_db(): print("查询数据库中的热点数据") time.sleep(0.5)return {"id": 1, "title": "热门商品", "price": 99.9}defget_hot_data(): cache_key = "cache:hot:product:1" lock_key = "lock:hot:product:1" cache_value = r.get(cache_key)if cache_value: print("命中热点缓存")return json.loads(cache_value) locked = r.set(lock_key, "1", nx=True, ex=10)if locked:try: print("获取互斥锁成功，回源数据库") data = query_hot_data_from_db() r.set(cache_key, json.dumps(data, ensure_ascii=False), ex=300)return datafinally: r.delete(lock_key) print("释放互斥锁")else: print("其他线程正在重建缓存，稍后重试") time.sleep(0.1) cache_value = r.get(cache_key)if cache_value:return json.loads(cache_value)returnNoneprint(get_hot_data())
输出：
获取互斥锁成功，回源数据库查询数据库中的热点数据释放互斥锁{'id': 1, 'title': '热门商品', 'price': 99.9}
3. 缓存雪崩
大量 key 同时过期，请求全部打到数据库。解决方式：过期时间加随机值、分批刷新、本地缓存、限流降级、Redis 高可用。
过期时间增加随机值：
import randomimport redisimport jsonr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)data = {"name": "张三"}base_ttl = 300random_ttl = random.randint(0, 60)ttl = base_ttl + random_ttlr.set("cache:user:1001", json.dumps(data, ensure_ascii=False), ex=ttl)print("缓存写入成功，过期时间:", ttl)
输出示例：
十六、Redis 限流实战
需求：同一个用户 60 秒内最多请求 5 次。
import redisr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)defcheck_rate_limit(user_id: int): key = f"rate_limit:user:{user_id}" count = r.incr(key)if count == 1: r.expire(key, 60) ttl = r.ttl(key)if count > 5:returnFalse, count, ttlreturnTrue, count, ttlfor i in range(1, 8): allowed, count, ttl = check_rate_limit(1001)if allowed: print(f"第 {i} 次请求：允许，当前次数={count}，剩余时间={ttl}s")else: print(f"第 {i} 次请求：拒绝，当前次数={count}，剩余时间={ttl}s")
输出：
第1次请求：允许，当前次数=1，剩余时间=60s第2次请求：允许，当前次数=2，剩余时间=60s...第6次请求：拒绝，当前次数=6，剩余时间=60s第7次请求：拒绝，当前次数=7，剩余时间=60s
此为固定窗口限流，更精细可用滑动窗口、令牌桶等算法。
十七、Redis 连接池封装
生产项目建议统一封装连接池，避免到处创建连接。
import redisclassRedisClient: _pool = None @classmethoddefget_client(cls):if cls._pool isNone: cls._pool = redis.ConnectionPool( host="localhost", port=6379, db=0, decode_responses=True, max_connections=20 )return redis.Redis(connection_pool=cls._pool)if __name__ == "__main__": r = RedisClient.get_client() r.set("app:name", "redis-demo") print(r.get("app:name"))
输出：
十八、FastAPI 中使用 Redis
1. 安装依赖
pip install fastapi uvicorn redis
2. 示例代码
from fastapi import FastAPIimport redisimport jsonapp = FastAPI()r = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)USER_DB = {1001: {"id": 1001, "name": "张三", "city": "深圳"},1002: {"id": 1002, "name": "李四", "city": "广州"},}@app.get("/users/{user_id}")defget_user(user_id: int): cache_key = f"cache:user:{user_id}" cache_value = r.get(cache_key)if cache_value:return {"source": "redis", "data": json.loads(cache_value)} user = USER_DB.get(user_id)if user isNone:return {"source": "db", "data": None, "message": "用户不存在"} r.set(cache_key, json.dumps(user, ensure_ascii=False), ex=300)return {"source": "db", "data": user}
启动：
uvicornmain:app--reload
第一次请求 curlhttp://127.0.0.1:8000/users/1001 返回 "source": "db"，第二次返回 "source": "redis"。
十九、Django 中使用 Redis 的思路
可直接封装 redis-py：
# utils/redis_client.pyimportredisredis_client = redis.Redis(host="localhost",port=6379,db=0,decode_responses=True)
业务中使用：
from utils.redis_client import redis_clientdefget_verify_code(phone): key = f"sms:code:{phone}"return redis_client.get(key)
若需全站缓存、session 缓存，可考虑 Django 的缓存框架并配置 Redis 后端。
二十、常见异常处理
1. 连接失败
importredisr = redis.Redis(host="localhost",port=6378,db=0,decode_responses=True,socket_connect_timeout=2)try:print(r.ping())exceptredis.exceptions.ConnectionError as e:print("Redis连接失败:", e)
2. 命令执行超时
r = redis.Redis(host="localhost",port=6379,db=0,decode_responses=True,socket_timeout=1)try:r.get("test")exceptredis.exceptions.TimeoutError as e:print("Redis操作超时:", e)
3. 类型错误
r.set("name", "张三")try: r.hset("name", "age", "18")except redis.exceptions.ResponseError as e: print("Redis 类型错误:", e)
输出：WRONGTYPE Operation against a key holding the wrong kind of value。
二十一、Redis Key 命名规范
建议使用风格：业务模块:业务含义:唯一标识
例如：
user:1001user:1001:profilesms:code:13800138000cache:product:2001lock:order:submit:3001rate_limit:user:1001task:queue:emailgame:rank:season:2026
避免无意义的 name、data、info 等，项目变大后难以维护。
二十二、Redis 使用原则
- 不要把 Redis 当永久数据库
- 所有缓存都应考虑过期策略
- 谨慎处理大 key
- 慎用 keys *
- cursor = 0 while True: cursor, keys = r.scan(cursor=cursor, match="cache:user:*", count=100) for key in keys: print(key) if cursor == 0: break
- 注意缓存与数据库一致性
- def update_user_name(user_id, new_name): # update db set name = new_name where id = user_id cache_key = f"cache:user:{user_id}" r.delete(cache_key)
二十三、完整实战：商品详情缓存
整合了缓存命中、空值缓存、随机过期时间的完整示例：
import redisimport jsonimport randomimport timefrom typing import Optional, Dict, Anyr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)PRODUCT_DB = {2001: {"id": 2001, "name": "无线鼠标", "price": 59.9, "stock": 120},2002: {"id": 2002, "name": "机械键盘", "price": 299.0, "stock": 35}}defquery_product_from_db(product_id: int) -> Optional[Dict[str, Any]]: print(f"[DB] 查询商品，product_id={product_id}") time.sleep(0.2)return PRODUCT_DB.get(product_id)defget_product(product_id: int) -> Optional[Dict[str, Any]]: cache_key = f"cache:product:{product_id}" cache_value = r.get(cache_key)if cache_value isnotNone:if cache_value == "": print("[Redis] 命中空值缓存")returnNone print("[Redis] 命中商品缓存")return json.loads(cache_value) print("[Redis] 未命中商品缓存") product = query_product_from_db(product_id)if product isNone: r.set(cache_key, "", ex=60) print("[Redis] 写入空值缓存，过期时间 60 秒")returnNone ttl = 300 + random.randint(0, 60) r.set(cache_key, json.dumps(product, ensure_ascii=False), ex=ttl) print(f"[Redis] 写入商品缓存，过期时间 {ttl} 秒")return productif __name__ == "__main__": print("第一次查询商品 2001") print(get_product(2001)) print("\n第二次查询商品 2001") print(get_product(2001)) print("\n查询不存在的商品 9999") print(get_product(9999)) print("\n再次查询不存在的商品 9999") print(get_product(9999))
二十四、完整实战：接口幂等控制
基于 Redis 的 set nx 实现唯一请求 ID 的幂等：
import redisimport timer = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)defsubmit_order(user_id: int, request_id: str): key = f"idempotent:order:{user_id}:{request_id}" success = r.set(key, "processing", nx=True, ex=60)ifnot success: print("重复请求，拒绝处理")return {"success": False, "message": "请勿重复提交"}try: print("首次请求，开始创建订单") time.sleep(0.5) order_id = "O202605100001" r.set(key, f"success:{order_id}", ex=300) print("订单创建成功:", order_id)return {"success": True, "order_id": order_id}except Exception as e: r.delete(key)raise eprint(submit_order(1001, "REQ-ABC-001"))print(submit_order(1001, "REQ-ABC-001"))
二十五、完整实战：验证码发送限制
60 秒内限制发送频率，验证码 5 分钟有效。
import redisimport randomr = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True)defsend_sms_code(phone: str): limit_key = f"sms:limit:{phone}" code_key = f"sms:code:{phone}" limited = r.set(limit_key, "1", nx=True, ex=60)ifnot limited: ttl = r.ttl(limit_key)return {"success": False, "message": f"发送太频繁，请 {ttl} 秒后再试"} code = str(random.randint(100000, 999999)) r.set(code_key, code, ex=300) print(f"模拟发送短信：手机号={phone}，验证码={code}")return {"success": True, "message": "验证码发送成功"}defverify_sms_code(phone: str, input_code: str): code_key = f"sms:code:{phone}" real_code = r.get(code_key)if real_code isNone:return {"success": False, "message": "验证码不存在或已过期"}if real_code != input_code:return {"success": False, "message": "验证码错误"} r.delete(code_key)return {"success": True, "message": "验证码校验成功"}print(send_sms_code("13800138000"))print(send_sms_code("13800138000"))
二十六、生产环境建议
- 密码不要硬编码
- import os import redis r = redis.Redis( host=os.getenv("REDIS_HOST", "localhost"), port=int(os.getenv("REDIS_PORT", "6379")), password=os.getenv("REDIS_PASSWORD"), db=int(os.getenv("REDIS_DB", "0")), decode_responses=True )
- 设置连接超时：socket_connect_timeout=3, socket_timeout=3，避免业务接口卡死。
- 重要接口要有降级逻辑：Redis 异常时降级查数据库，并包装 safe_get_cache 方法捕获 RedisError。
- 缓存数据必须可重建
- 监控 Redis：关注内存、连接数、慢查询、命中率、主从同步状态等。
二十七、总结
Redis 在 Python 项目中非常实用，尤其适合缓存、验证码、Token、计数器、排行榜、限流、分布式锁、简单队列、发布订阅、接口幂等等场景。实际使用时需注意：
一句话总结：Redis 很快，但快不是乱用的理由。用得好，它是系统性能加速器；用不好，它就是凌晨三点叫醒你的“红色报警器”。
