Redis 缓存设计与优化
缓存是提升系统性能的关键手段。Redis 作为高性能缓存系统，广泛应用于各种场景。本文将深入缓存设计模式，解析常见问题及解决方案。
一、缓存模式
1.1 Cache-Aside（旁路缓存）
最常用的缓存模式：
读操作：
1. 先读缓存
2. 缓存未命中，读数据库
3. 写入缓存
4. 返回结果
写操作：
1. 更新数据库
2. 删除缓存代码实现：
def get_user(user_id):
# 1. 读缓存
key = f"user:{user_id}"
user = redis.get(key)
if user:
return deserialize(user)
# 2. 读数据库
user = db.query("SELECT * FROM users WHERE id = ?", user_id)
if user:
# 3. 写缓存
redis.setex(key, 300, serialize(user))
return user
def update_user(user_id, data):
# 1. 更新数据库
db.execute("UPDATE users SET ? WHERE id = ?", data, user_id)
# 2. 删除缓存
redis.delete(f"user:{user_id}")优点：
- 简单易懂
- 缓存只存储热点数据
缺点：
- 存在并发一致性问题
1.2 Read/Write Through
读操作：
1. 应用读缓存
2. 缓存未命中，缓存系统读数据库
3. 缓存系统返回
写操作：
1. 应用写缓存
2. 缓存系统异步写数据库特点：
- 缓存负责与数据库同步
- 应用无需关心缓存逻辑
1.3 Write Behind
写操作：
1. 应用只写缓存
2. 缓存异步批量写数据库特点：
- 高性能
- 可能丢失数据
二、缓存问题与解决方案
2.1 缓存穿透
问题描述：
查询不存在的数据
↓
缓存未命中
↓
查询数据库（也未命中）
↓
每次请求都打到数据库
攻击场景：
恶意请求大量不存在的 key
导致数据库压力过大解决方案：
方案 1：布隆过滤器
from redis.commands.search.field import TextField
from redis.commands.search.indexDefinition import IndexDefinition
from redisbloom.client import Client
# 初始化布隆过滤器
bf = Client(redis)
# 添加元素
bf.add("bloom_users", "user_1001")
bf.add("bloom_users", "user_1002")
# 检查是否存在
if bf.exists("bloom_users", "user_1001"):
# 可能存在，查询缓存/数据库
user = get_user(1001)
else:
# 一定不存在，直接返回
return None方案 2：缓存空值
def get_user(user_id):
key = f"user:{user_id}"
# 读缓存
user = redis.get(key)
if user == "__NULL__":
return None
if user:
return deserialize(user)
# 读数据库
user = db.query("SELECT * FROM users WHERE id = ?", user_id)
if not user:
# 缓存空值，防止穿透
redis.setex(key, 60, "__NULL__")
return None
redis.setex(key, 300, serialize(user))
return user方案 3：参数校验
def get_user(user_id):
# 参数校验
if not user_id or user_id <= 0:
return None
# 正常查询
return query_user(user_id)2.2 缓存击穿
问题描述：
热点 key 过期
↓
大量并发请求
↓
全部打到数据库
↓
数据库崩溃解决方案：
方案 1：互斥锁
def get_hot_data(key):
data = redis.get(key)
if data:
return deserialize(data)
# 获取互斥锁
lock_key = f"lock:{key}"
if redis.set(lock_key, "1", nx=True, ex=10):
try:
# 双重检查
data = redis.get(key)
if data:
return deserialize(data)
# 查询数据库
data = query_db(key)
# 写入缓存
redis.setex(key, 300, serialize(data))
return data
finally:
redis.delete(lock_key)
else:
# 等待后重试
time.sleep(0.1)
return get_hot_data(key)方案 2：永不过期
# 热点数据不设置过期时间
redis.set("hot_key", serialize(data))
# 异步更新
def async_update():
while True:
data = query_db("hot_key")
redis.set("hot_key", serialize(data))
time.sleep(300) # 5 分钟更新一次方案 3：逻辑过期
def get_data_with_logic_expire(key):
data = redis.get(key)
if data:
data_obj = deserialize(data)
# 检查逻辑过期时间
if data_obj['expire_time'] > time.time():
return data_obj['data']
else:
# 异步更新
if redis.set(f"lock:{key}", "1", nx=True, ex=10):
threading.Thread(target=update_data, args=(key,)).start()
return data_obj['data']
# 首次查询
return query_and_cache(key)
def update_data(key):
data = query_db(key)
cached = {
'data': data,
'expire_time': time.time() + 300
}
redis.setex(key, 3600, serialize(cached)) # 物理过期 1 小时2.3 缓存雪崩
问题描述：
大量 key 同时过期
↓
请求全部打到数据库
↓
数据库崩溃原因：
- 缓存集中过期
- Redis 宕机
解决方案：
方案 1：随机过期时间
def set_with_random_expire(key, data, base_expire):
# 基础时间 + 随机时间
expire = base_expire + random.randint(0, 300)
redis.setex(key, expire, serialize(data))
# 使用
set_with_random_expire("user:1001", user, 300)
set_with_random_expire("user:1002", user, 300)
# 实际过期时间：300-600 秒方案 2：多级缓存
def get_data(key):
# L1 缓存（本地内存）
data = local_cache.get(key)
if data:
return data
# L2 缓存（Redis）
data = redis.get(key)
if data:
local_cache.set(key, data, 60) # 本地缓存 1 分钟
return deserialize(data)
# 数据库
data = query_db(key)
redis.setex(key, 300, serialize(data))
local_cache.set(key, data, 60)
return data方案 3：高可用架构
Redis Cluster + Sentinel
↓
自动故障转移
↓
避免单点故障三、缓存一致性
3.1 一致性方案
方案 1：先更新数据库，再删除缓存
def update_user(user_id, data):
db.execute("UPDATE users SET ? WHERE id = ?", data, user_id)
redis.delete(f"user:{user_id}")问题：
线程 A: 更新数据库
线程 B: 读缓存（旧数据）
线程 B: 删除缓存
线程 A: 删除缓存
结果：缓存为空，下次查询会加载旧数据方案 2：延时双删
def update_user(user_id, data):
# 1. 删除缓存
redis.delete(f"user:{user_id}")
# 2. 更新数据库
db.execute("UPDATE users SET ? WHERE id = ?", data, user_id)
# 3. 延迟后再次删除
time.sleep(0.5)
redis.delete(f"user:{user_id}")方案 3：监听 binlog
# 使用 Canal 监听 MySQL binlog
def on_binlog_event(event):
if event.table == 'users':
if event.type == 'UPDATE':
redis.delete(f"user:{event.user_id}")方案 4：分布式事务
# 使用事务保证原子性
def update_user(user_id, data):
with transaction():
db.execute("UPDATE users SET ? WHERE id = ?", data, user_id)
redis.delete(f"user:{user_id}")3.2 一致性级别选择
| 场景 | 一致性要求 | 方案 |
|---|---|---|
| 金融交易 | 强一致 | 分布式事务 |
| 用户信息 | 最终一致 | 先 DB 后删缓存 |
| 商品库存 | 最终一致 | 延时双删 |
| 热点新闻 | 弱一致 | 异步更新 |
四、缓存优化
4.1 数据结构优化
# ❌ 不推荐：大 Key
user = {
'id': 1001,
'name': 'John',
'profile': {...}, # 大对象
'orders': [...] # 大数组
}
redis.set("user:1001", json.dumps(user))
# ✅ 推荐：拆分存储
redis.hset("user:1001:base", mapping={
'id': 1001,
'name': 'John'
})
redis.hset("user:1001:profile", mapping=profile)
redis.lpush("user:1001:orders", *orders)4.2 批量操作
# ❌ 不推荐：逐个操作
for user_id in user_ids:
redis.get(f"user:{user_id}")
# ✅ 推荐：批量操作
pipe = redis.pipeline()
for user_id in user_ids:
pipe.get(f"user:{user_id}")
results = pipe.execute()4.3 内存优化
# 1. 使用合适的数据结构
# Hash vs String
redis.hset("user:1001", "name", "John") # 更省内存
# 2. 设置过期时间
redis.setex("temp_key", 300, value)
# 3. 使用内存淘汰策略
# maxmemory-policy allkeys-lru
# 4. 定期清理
redis.scan_iter("temp:*") # 扫描临时 key
redis.delete("temp:key") # 删除4.4 监控指标
# 关键指标
INFO stats
- keyspace_hits # 缓存命中数
- keyspace_misses # 缓存未命中数
- hit_rate = hits / (hits + misses) # 命中率
INFO memory
- used_memory # 已用内存
- maxmemory # 最大内存
- mem_fragmentation # 碎片率
INFO keyspace
- db0:keys=1000,expires=500 # key 数量、过期数
# 告警
if hit_rate < 0.8:
alert("缓存命中率过低")
if used_memory / maxmemory > 0.9:
alert("内存使用率过高")五、最佳实践总结
5.1 设计原则
1. 只缓存热点数据
2. 设置合理的过期时间
3. 避免大 Key
4. 使用批量操作
5. 监控命中率5.2 问题处理
| 问题 | 检测 | 解决 |
|---|---|---|
| 穿透 | 大量 key 未命中 | 布隆过滤器、缓存空值 |
| 击穿 | 热点 key 过期 | 互斥锁、永不过期 |
| 雪崩 | 大量 key 同时过期 | 随机过期、多级缓存 |
| 不一致 | 缓存与 DB 数据不同 | 延时双删、监听 binlog |
5.3 配置建议
# 内存配置
maxmemory 4gb
maxmemory-policy allkeys-lru
# 持久化
save 900 1
appendonly yes
# 网络
timeout 300
tcp-keepalive 60总结
Redis 缓存设计核心要点：
| 模式 | 适用场景 | 一致性 |
|---|---|---|
| Cache-Aside | 通用场景 | 最终一致 |
| Read Through | 读多写少 | 最终一致 |
| Write Behind | 高性能场景 | 弱一致 |
问题解决方案：
- 穿透 → 布隆过滤器
- 击穿 → 互斥锁
- 雪崩 → 随机过期
- 不一致 → 延时双删
掌握缓存设计，构建高性能系统！