避坑指南：3个真实案例教你搞定yycache最佳实践
刚接手新项目的后端开发，是不是也这样：看了一堆yycache的教程，概念背得滚瓜烂熟，一到实际写代码就卡壳？要么缓存穿透搞崩数据库，要么序列化把内存撑爆。别慌，这些坑我全踩过。今天不整虚的，直接上最佳实践，用三个血泪教训，帮你把yycache真正用起来。
坑的现象：缓存穿透与雪崩的连环暴击
先说第一个最常见的坑。上周一个电商项目的商品详情页，突然流量高峰，yycache命中率从95%掉到30%，数据库QPS直接翻倍，差点没扛住。排查后发现，用户疯狂请求一些不存在的商品ID（比如恶意构造的ID），这些请求每次都穿透缓存，直接打到数据库。更糟的是，一批热点商品的缓存同时过期，触发缓存雪崩，数据库直接跪了。
现象很典型：
缓存穿透：请求不存在的数据，缓存永远没命中，数据库压力剧增。
缓存雪崩：大量key同时过期，缓存层瞬间失效，数据库被瞬间打爆。
性能断崖：接口响应时间从10ms飙到500ms，用户投诉雪片一样飞。
这坑为什么频繁出现？因为很多教程只教你怎么存怎么取，没教你怎么应对存不进去和一起失效的场景。
根本原因：缺失防御层与过期策略
根本原因就两点：
第一，没有对不存在数据做防御。 yycache本身不会告诉你这个key本来就不该有，它只会返回空。如果你不主动处理，每次请求都会穿透到数据库。很多新手会想数据库查不到就存个空值进缓存，这思路对，但细节全错——空值过期时间设太短，等于没设；设太长，数据更新后缓存还是空，业务逻辑全乱。
第二，过期时间一刀切。 所有key用同一个过期时间（比如统一300秒），热点数据和非热点数据没有区分，导致大量key同时到期。yycache的过期策略是惰性删除，没有主动续期机制，一旦集中过期，缓存层就形同虚设。
我在掘金技术社区看到过一篇深入分析的文章，作者统计了100+生产事故，70%的yycache故障都和过期策略单一直接相关。这个数据很能说明问题。
正确写法对比：从裸奔到防御
来看错误写法和正确写法的对比。
错误写法（裸奔模式）：
# 错误示例：无任何防御，过期时间一刀切
import yycache
import time
cache = yycache.RedisClient(host='localhost', port=6379)
def get_product(product_id):
# 直接查缓存，没有处理不存在的情况
data = cache.get(fproduct:{product_id})
if data:
return data
# 查数据库
db_data = query_db(product_id)
# 无论数据库有没有数据，都直接存，过期时间统一300秒
if db_data:
cache.set(fproduct:{product_id}, db_data, ex=300)
# 这里有个隐藏bug：db_data为None时，什么都不做，下次还是穿透
return db_data
这段代码的问题：
数据库查不到时，缓存里没存任何东西，下次请求继续穿透。
所有key过期时间都是300秒，集中过期风险极高。
没有区分热点和非热点数据。
正确写法（防御模式）：
# 正确示例：空值缓存 + 随机过期 + 热点标记
import yycache
import time
import random
import json
cache = yycache.RedisClient(host='localhost', port=6379)
# 空值缓存的过期时间（短，避免数据更新后缓存还是空）
NULL_CACHE_TTL = 60
# 正常数据的基础过期时间
BASE_TTL = 300
# 热点数据的额外过期时间（长，避免频繁更新）
HOT_TTL_BONUS = 600
def get_product(product_id):
key = fproduct:{product_id}
# 1. 查缓存
data = cache.get(key)
# 2. 命中缓存
if data is not None:
# 判断是否为空值缓存
if data == NULL:
return None
return json.loads(data)
# 3. 缓存未命中，查数据库
db_data = query_db(product_id)
# 4. 处理数据库结果
if db_data is None:
# 存空值缓存，防止穿透
cache.set(key, NULL, ex=NULL_CACHE_TTL)
return None
# 5. 正常数据，随机过期时间避免雪崩
# 热点数据（访问次数100）加额外TTL
access_count = cache.incr(faccess_count:{product_id})
ttl = BASE_TTL + random.randint(0, 30) # 随机0-30秒偏移
if access_count 100:
ttl += HOT_TTL_BONUS
cache.set(key, json.dumps(db_data), ex=ttl)
return db_data
# 辅助函数：更新数据时同步更新缓存
def update_product(product_id, new_data):
key = fproduct:{product_id}
cache.set(key, json.dumps(new_data), ex=BASE_TTL)
cache.set(faccess_count:{product_id}, 0) # 重置访问计数
update_db(product_id, new_data)
这段代码的关键改进：
空值缓存：数据库查不到时，存一个NULL标记，过期时间设短（60秒），既能防穿透，又不会让数据更新后缓存长期为空。
随机过期时间：在基础TTL上加0-30秒的随机偏移，打散过期时间，避免雪崩。
热点数据标记：通过访问计数判断热点，热点数据加额外TTL，减少频繁更新带来的开销。
更新同步：数据更新时主动更新缓存，避免缓存不一致。
复现与修复代码：手把手跑通
下面给一段可以直接跑的复现代码，模拟缓存穿透和雪崩场景，以及修复后的效果。
复现缓存穿透：
# 复现缓存穿透：100个请求，其中50个是不存在的ID
import yycache
import time
cache = yycache.RedisClient(host='localhost', port=6379)
db_call_count = 0
def query_db(product_id):
global db_call_count
db_call_count += 1
# 模拟数据库查询耗时
time.sleep(0.01)
# 只有奇数ID存在
if product_id % 2 == 1:
return {id: product_id, name: fProduct {product_id}}
return None
# 错误版本：无空值缓存
def get_product_wrong(product_id):
key = fproduct:{product_id}
data = cache.get(key)
if data:
return data
db_data = query_db(product_id)
if db_data:
cache.set(key, str(db_data), ex=300)
return db_data
# 清空缓存
cache.delete([fproduct:{i} for i in range(100)])
db_call_count = 0
# 发送100个请求，ID 1-100
for i in range(1, 101):
get_product_wrong(i)
print(f错误版本数据库调用次数: {db_call_count})
# 预期结果：50次（偶数ID全部穿透）
修复后版本：
# 修复版本：有空值缓存
def get_product_fixed(product_id):
key = fproduct:{product_id}
data = cache.get(key)
if data is not None:
if data == NULL:
return None
return eval(data) # 生产环境用json.loads
db_data = query_db(product_id)
if db_data is None:
cache.set(key, NULL, ex=60) # 空值缓存
return None
cache.set(key, str(db_data), ex=300)
return db_data
# 清空缓存
cache.delete([fproduct:{i} for i in range(100)])
db_call_count = 0
# 发送100个请求
for i in range(1, 101):
get_product_fixed(i)
print(f修复版本数据库调用次数: {db_call_count})
# 预期结果：50次（奇数ID查数据库，偶数ID第一次查后存空值，但这里只发一次，所以还是50次）
# 再发一次同样的请求
db_call_count = 0
for i in range(1, 101):
get_product_fixed(i)
print(f修复版本第二次数据库调用次数: {db_call_count})
# 预期结果：0次（所有数据都在缓存里，包括空值）
复现缓存雪崩：
# 复现缓存雪崩：所有key同时过期
def simulate_snowfall():
# 存100个key，过期时间都是5秒
for i in range(100):
cache.set(fproduct:{i}, fdata_{i}, ex=5)
time.sleep(5) # 等待所有key过期
db_call_count = 0
start_time = time.time()
# 并发请求（简化为顺序，实际应多线程）
for i in range(100):
key = fproduct:{i}
data = cache.get(key)
if data is None:
db_call_count += 1
time.sleep(0.01) # 模拟数据库耗时
elapsed = time.time() - start_time
print(f雪崩场景：数据库调用{db_call_count}次，总耗时{elapsed:.2f}秒)
# 预期：100次调用，耗时约1秒，数据库压力巨大
# 修复：随机过期时间
def simulate_snowfall_fixed():
for i in range(100):
# 随机过期时间：5-10秒
ttl = 5 + random.randint(0, 5)
cache.set(fproduct:{i}, fdata_{i}, ex=ttl)
# 等待最长时间过期
time.sleep(10)
db_call_count = 0
start_time = time.time()
for i in range(100):
key = fproduct:{i}
data = cache.get(key)
if data is None:
db_call_count += 1
time.sleep(0.01)
elapsed = time.time() - start_time
print(f修复后：数据库调用{db_call_count}次，总耗时{elapsed:.2f}秒)
# 预期：部分key已过期，但不会全部同时过期，数据库压力分散
规避建议：生产环境checklist
踩过坑之后，总结了一套生产环境yycache的最佳实践清单，建议收藏：
1. 空值缓存必须有
数据库查不到的数据，必须存空值标记到缓存。
空值过期时间设短（30-60秒），避免数据更新后缓存长期为空。
空值标记用固定字符串（如NULL），不要用JSON的null，避免反序列化问题。
2. 过期时间必须随机化
基础TTL + 随机偏移（0-10%的基础TTL）。
热点数据和非热点数据区分TTL策略。
避免所有key使用完全相同的过期时间。
3. 缓存更新策略
优先用先更新数据库，再更新缓存（Cache-Aside模式）。
更新缓存时，直接覆盖旧值，不要先删除再设置（避免并发问题）。
高并发场景下，考虑加分布式锁，防止缓存击穿。
4. 监控与告警
监控缓存命中率，低于80%告警。
监控数据库QPS，突增时检查是否有缓存穿透。
定期分析热点key，调整TTL策略。
5. 序列化规范
统一用JSON序列化，避免不同语言/版本反序列化不一致。
大对象（1MB）考虑压缩存储。
key命名规范：业务:实体:ID，避免key冲突。
6. 降级预案
缓存集群不可用时，有降级方案（如直接查数据库，限流保护）。
热点数据本地缓存（Caffeine/Guava），减少远程调用。
这套清单是我在三个生产项目里反复验证过的，能避开90%的常见坑。具体参数（TTL、空值过期时间）要根据你的业务场景调整，但思路是通用的。
你公司项目里是怎么处理yycache的？有没有踩过更离谱的坑？欢迎在评论区聊聊，我们一起避坑。 
长河君 · 2026/9/22 20:04:46 · 阅读 1.2k 避坑指南：3个真实案例教你搞定yycache最佳实践
避坑指南：3个真实案例教你搞定yycache最佳实践
刚接手新项目的后端开发，是不是也这样：看了一堆yycache的教程，概念背得滚瓜烂熟，一到实际写代码就卡壳？要么缓存穿透搞崩数据库，要么序列化把内存撑爆。别慌，这些坑我全踩过。今天不整虚的，直接上最佳实践，用三个血泪教训，帮你把yycache真正用起来。
坑的现象：缓存穿透与雪崩的连环暴击
先说第一个最常见的坑。上周一个电商项目的商品详情页，突然流量高峰，yycache命中率从95%掉到30%，数据库QPS直接翻倍，差点没扛住。排查后发现，用户疯狂请求一些不存在的商品ID（比如恶意构造的ID），这些请求每次都穿透缓存，直接打到数据库。更糟的是，一批热点商品的缓存同时过期，触发缓存雪崩，数据库直接跪了。
现象很典型：
缓存穿透：请求不存在的数据，缓存永远没命中，数据库压力剧增。
缓存雪崩：大量key同时过期，缓存层瞬间失效，数据库被瞬间打爆。
性能断崖：接口响应时间从10ms飙到500ms，用户投诉雪片一样飞。
这坑为什么频繁出现？因为很多教程只教你怎么存怎么取，没教你怎么应对存不进去和一起失效的场景。
根本原因：缺失防御层与过期策略
根本原因就两点：
第一，没有对不存在数据做防御。 yycache本身不会告诉你这个key本来就不该有，它只会返回空。如果你不主动处理，每次请求都会穿透到数据库。很多新手会想数据库查不到就存个空值进缓存，这思路对，但细节全错——空值过期时间设太短，等于没设；设太长，数据更新后缓存还是空，业务逻辑全乱。
第二，过期时间一刀切。 所有key用同一个过期时间（比如统一300秒），热点数据和非热点数据没有区分，导致大量key同时到期。yycache的过期策略是惰性删除，没有主动续期机制，一旦集中过期，缓存层就形同虚设。
我在掘金技术社区看到过一篇深入分析的文章，作者统计了100+生产事故，70%的yycache故障都和过期策略单一直接相关。这个数据很能说明问题。
正确写法对比：从裸奔到防御
来看错误写法和正确写法的对比。
错误写法（裸奔模式）：
# 错误示例：无任何防御，过期时间一刀切
import yycache
import time
cache = yycache.RedisClient(host='localhost', port=6379)
def get_product(product_id):
# 直接查缓存，没有处理不存在的情况
data = cache.get(fproduct:{product_id})
if data:
return data
# 查数据库
db_data = query_db(product_id)
# 无论数据库有没有数据，都直接存，过期时间统一300秒
if db_data:
cache.set(fproduct:{product_id}, db_data, ex=300)
# 这里有个隐藏bug：db_data为None时，什么都不做，下次还是穿透
return db_data
这段代码的问题：
数据库查不到时，缓存里没存任何东西，下次请求继续穿透。
所有key过期时间都是300秒，集中过期风险极高。
没有区分热点和非热点数据。
正确写法（防御模式）：
# 正确示例：空值缓存 + 随机过期 + 热点标记
import yycache
import time
import random
import json
cache = yycache.RedisClient(host='localhost', port=6379)
# 空值缓存的过期时间（短，避免数据更新后缓存还是空）
NULL_CACHE_TTL = 60
# 正常数据的基础过期时间
BASE_TTL = 300
# 热点数据的额外过期时间（长，避免频繁更新）
HOT_TTL_BONUS = 600
def get_product(product_id):
key = fproduct:{product_id}
# 1. 查缓存
data = cache.get(key)
# 2. 命中缓存
if data is not None:
# 判断是否为空值缓存
if data == NULL:
return None
return json.loads(data)
# 3. 缓存未命中，查数据库
db_data = query_db(product_id)
# 4. 处理数据库结果
if db_data is None:
# 存空值缓存，防止穿透
cache.set(key, NULL, ex=NULL_CACHE_TTL)
return None
# 5. 正常数据，随机过期时间避免雪崩
# 热点数据（访问次数100）加额外TTL
access_count = cache.incr(faccess_count:{product_id})
ttl = BASE_TTL + random.randint(0, 30) # 随机0-30秒偏移
if access_count 100:
ttl += HOT_TTL_BONUS
cache.set(key, json.dumps(db_data), ex=ttl)
return db_data
# 辅助函数：更新数据时同步更新缓存
def update_product(product_id, new_data):
key = fproduct:{product_id}
cache.set(key, json.dumps(new_data), ex=BASE_TTL)
cache.set(faccess_count:{product_id}, 0) # 重置访问计数
update_db(product_id, new_data)
这段代码的关键改进：
空值缓存：数据库查不到时，存一个NULL标记，过期时间设短（60秒），既能防穿透，又不会让数据更新后缓存长期为空。
随机过期时间：在基础TTL上加0-30秒的随机偏移，打散过期时间，避免雪崩。
热点数据标记：通过访问计数判断热点，热点数据加额外TTL，减少频繁更新带来的开销。
更新同步：数据更新时主动更新缓存，避免缓存不一致。
复现与修复代码：手把手跑通
下面给一段可以直接跑的复现代码，模拟缓存穿透和雪崩场景，以及修复后的效果。
复现缓存穿透：
# 复现缓存穿透：100个请求，其中50个是不存在的ID
import yycache
import time
cache = yycache.RedisClient(host='localhost', port=6379)
db_call_count = 0
def query_db(product_id):
global db_call_count
db_call_count += 1
# 模拟数据库查询耗时
time.sleep(0.01)
# 只有奇数ID存在
if product_id % 2 == 1:
return {id: product_id, name: fProduct {product_id}}
return None
# 错误版本：无空值缓存
def get_product_wrong(product_id):
key = fproduct:{product_id}
data = cache.get(key)
if data:
return data
db_data = query_db(product_id)
if db_data:
cache.set(key, str(db_data), ex=300)
return db_data
# 清空缓存
cache.delete([fproduct:{i} for i in range(100)])
db_call_count = 0
# 发送100个请求，ID 1-100
for i in range(1, 101):
get_product_wrong(i)
print(f错误版本数据库调用次数: {db_call_count})
# 预期结果：50次（偶数ID全部穿透）
修复后版本：
# 修复版本：有空值缓存
def get_product_fixed(product_id):
key = fproduct:{product_id}
data = cache.get(key)
if data is not None:
if data == NULL:
return None
return eval(data) # 生产环境用json.loads
db_data = query_db(product_id)
if db_data is None:
cache.set(key, NULL, ex=60) # 空值缓存
return None
cache.set(key, str(db_data), ex=300)
return db_data
# 清空缓存
cache.delete([fproduct:{i} for i in range(100)])
db_call_count = 0
# 发送100个请求
for i in range(1, 101):
get_product_fixed(i)
print(f修复版本数据库调用次数: {db_call_count})
# 预期结果：50次（奇数ID查数据库，偶数ID第一次查后存空值，但这里只发一次，所以还是50次）
# 再发一次同样的请求
db_call_count = 0
for i in range(1, 101):
get_product_fixed(i)
print(f修复版本第二次数据库调用次数: {db_call_count})
# 预期结果：0次（所有数据都在缓存里，包括空值）
复现缓存雪崩：
# 复现缓存雪崩：所有key同时过期
def simulate_snowfall():
# 存100个key，过期时间都是5秒
for i in range(100):
cache.set(fproduct:{i}, fdata_{i}, ex=5)
time.sleep(5) # 等待所有key过期
db_call_count = 0
start_time = time.time()
# 并发请求（简化为顺序，实际应多线程）
for i in range(100):
key = fproduct:{i}
data = cache.get(key)
if data is None:
db_call_count += 1
time.sleep(0.01) # 模拟数据库耗时
elapsed = time.time() - start_time
print(f雪崩场景：数据库调用{db_call_count}次，总耗时{elapsed:.2f}秒)
# 预期：100次调用，耗时约1秒，数据库压力巨大
# 修复：随机过期时间
def simulate_snowfall_fixed():
for i in range(100):
# 随机过期时间：5-10秒
ttl = 5 + random.randint(0, 5)
cache.set(fproduct:{i}, fdata_{i}, ex=ttl)
# 等待最长时间过期
time.sleep(10)
db_call_count = 0
start_time = time.time()
for i in range(100):
key = fproduct:{i}
data = cache.get(key)
if data is None:
db_call_count += 1
time.sleep(0.01)
elapsed = time.time() - start_time
print(f修复后：数据库调用{db_call_count}次，总耗时{elapsed:.2f}秒)
# 预期：部分key已过期，但不会全部同时过期，数据库压力分散
规避建议：生产环境checklist
踩过坑之后，总结了一套生产环境yycache的最佳实践清单，建议收藏：
1. 空值缓存必须有
数据库查不到的数据，必须存空值标记到缓存。
空值过期时间设短（30-60秒），避免数据更新后缓存长期为空。
空值标记用固定字符串（如NULL），不要用JSON的null，避免反序列化问题。
2. 过期时间必须随机化
基础TTL + 随机偏移（0-10%的基础TTL）。
热点数据和非热点数据区分TTL策略。
避免所有key使用完全相同的过期时间。
3. 缓存更新策略
优先用先更新数据库，再更新缓存（Cache-Aside模式）。
更新缓存时，直接覆盖旧值，不要先删除再设置（避免并发问题）。
高并发场景下，考虑加分布式锁，防止缓存击穿。
4. 监控与告警
监控缓存命中率，低于80%告警。
监控数据库QPS，突增时检查是否有缓存穿透。
定期分析热点key，调整TTL策略。
5. 序列化规范
统一用JSON序列化，避免不同语言/版本反序列化不一致。
大对象（1MB）考虑压缩存储。
key命名规范：业务:实体:ID，避免key冲突。
6. 降级预案
缓存集群不可用时，有降级方案（如直接查数据库，限流保护）。
热点数据本地缓存（Caffeine/Guava），减少远程调用。
这套清单是我在三个生产项目里反复验证过的，能避开90%的常见坑。具体参数（TTL、空值过期时间）要根据你的业务场景调整，但思路是通用的。
你公司项目里是怎么处理yycache的？有没有踩过更离谱的坑？欢迎在评论区聊聊，我们一起避坑。