Redis 缓存在面试题系统中的应用热点题目与过期策略一、面试题系统的读写特征面试题系统有很典型的读多写少特征——一套题上线后可能有数千人同时浏览但题目内容的变更频率很低可能几周才更新一次。这种场景是缓存大显身手的地方。但简单的查了缓存就返回没查到就查数据库策略在题量很大、题目热门程度差异大时会遇到问题热门题目如 Two Sum的缓存命中率很高冷门题目的缓存可能永远不被命中占着内存没产出。二、多级缓存架构flowchart TD A[用户请求题目] -- B[CDN 缓存: 静态页面] B --|过期/未命中| C[本地缓存: Caffeine/LRU] C --|未命中| D[Redis 集群] D --|未命中| E[数据库] D -- F[热点检测] F -- F1[滑动窗口计数器] F -- F2[Top-K 热点题目列表] F2 -- G[预加载到本地缓存]三、实现热点感知缓存系统import time import threading from collections import defaultdict, deque from typing import Optional, Any from dataclasses import dataclass dataclass class CacheStats: 缓存统计信息 hits: int 0 misses: int 0 total_requests: int 0 property def hit_rate(self) - float: if self.total_requests 0: return 0.0 return self.hits / self.total_requests class HotKeyDetector: 热点 Key 检测器 使用滑动窗口统计每个 key 的访问频率。 在固定窗口内访问次数超过阈值的 key 被标记为热点。 def __init__(self, window_seconds: int 60, hot_threshold: int 100): self.window window_seconds self.threshold hot_threshold # key → 访问时间戳队列 self._access_records: dict[str, deque] defaultdict(deque) self._hot_keys: set[str] set() self._lock threading.Lock() def record_access(self, key: str) - bool: 记录一次访问返回该 key 是否为热点 now time.time() with self._lock: records self._access_records[key] records.append(now) # 清理过期记录只清理当前 key 的避免全量扫描 while records and records[0] now - self.window: records.popleft() # 判断是否为热点 is_hot len(records) self.threshold if is_hot: self._hot_keys.add(key) elif key in self._hot_keys: self._hot_keys.discard(key) return is_hot def get_hot_keys(self) - set[str]: 获取当前热点 key 集合 with self._lock: return self._hot_keys.copy() def cleanup_expired(self) - None: 定期清理所有过期记录后台任务 now time.time() with self._lock: expired_keys [] for key, records in self._access_records.items(): while records and records[0] now - self.window: records.popleft() if not records: expired_keys.append(key) for key in expired_keys: del self._access_records[key] class SmartCache: 智能缓存根据访问热度自动调整缓存策略 策略 1. 热点题目缓存 TTL 长1 小时预加载到本地 2. 普通题目缓存 TTL 短5 分钟 3. 冷门题目不缓存或极短 TTL 1 分钟 这种差异化策略让有限的缓存空间服务于最常访问的题目。 TTL_HOT 3600 # 热点 1 小时 TTL_NORMAL 300 # 普通 5 分钟 TTL_COLD 60 # 冷门 1 分钟 def __init__(self, redis_client, local_cache_size: int 1000): self.redis redis_client self.detector HotKeyDetector() self.stats CacheStats() # 本地缓存LRU存热点数据减少网络 IO self._local_cache: dict[str, tuple[Any, float]] {} self._local_max_size local_cache_size def get_problem(self, problem_id: int) - Optional[dict]: 获取题目多层缓存查询 key fproblem:{problem_id} self.stats.total_requests 1 # 通知热度检测器 is_hot self.detector.record_access(key) # 第一层本地缓存仅热点数据驻留 if key in self._local_cache: value, expire_at self._local_cache[key] if time.time() expire_at: self.stats.hits 1 return value del self._local_cache[key] # 第二层Redis 缓存 cached self.redis.get(key) if cached: self.stats.hits 1 value self._deserialize(cached) # 热点数据回写到本地缓存 if is_hot: self._local_cache_set(key, value, self.TTL_HOT) return value # 第三层数据库查询 self.stats.misses 1 problem self._query_database(problem_id) if problem is None: # 缓存空值防止缓存穿透 self.redis.setex(f{key}:null, self.TTL_COLD, 1) return None # 写入缓存差异化 TTL ttl self.TTL_HOT if is_hot else ( self.TTL_NORMAL if problem.get(difficulty) ! Hard else self.TTL_COLD ) self.redis.setex(key, ttl, self._serialize(problem)) return problem def preload_hot_problems(self) - None: 预加载热点题目到本地缓存 在流量高峰前如每天早上 9 点调用 提前将预测的热点题目加载到本地缓存。 hot_keys self.detector.get_hot_keys() for key in hot_keys: if key.startswith(problem:): problem_id int(key.split(:)[1]) problem self.get_problem(problem_id) if problem: self._local_cache_set(key, problem, self.TTL_HOT) def _local_cache_set(self, key: str, value: Any, ttl: int) - None: 写入本地缓存超过容量时 LRU 淘汰 # 简易 LRU 淘汰删除最早的条目 if len(self._local_cache) self._local_max_size: oldest_key next(iter(self._local_cache)) del self._local_cache[oldest_key] self._local_cache[key] (value, time.time() ttl) def _query_database(self, problem_id: int) - Optional[dict]: 数据库查询 # 实际查询逻辑 return None staticmethod def _serialize(data: dict) - str: import json return json.dumps(data, ensure_asciiFalse) staticmethod def _deserialize(data: str) - dict: import json return json.loads(data)四、缓存策略的工程考量4.1 缓存穿透大量请求查询不存在的题目 ID如 ID-1由于数据库查不到缓存也不会写入导致每次请求都穿透到数据库。解决方案是缓存空值TTL 设短一些1-5 分钟。4.2 缓存雪崩大量缓存在同一时刻过期导致瞬时请求全部打到数据库。解决方案是TTL 加随机偏移如 300s random(0, 60)s。4.3 热点检测的滞后性滑动窗口检测有固有的滞后——key 成为热点后需要等待下一个窗口才能被识别。对于突发流量如某题目突然上了热搜可以考虑增加实时计数器作为补充。4.4 缓存与数据的最终一致性题目内容的变更不是实时的但需要在可接受的时间窗口如 5 分钟内生效。缓存失效策略主动失效 TTL 兜底是这一问题的标准解决方案。五、总结面试题系统的缓存设计不是加缓存就完事了。热冷数据的差异性、多层缓存的协同、缓存穿透和雪崩的防护每一项都需要针对场景特点做出决策。核心思路是让有限的缓存资源尽可能服务于高频访问的数据——这就是热点检测和差异化 TTL 策略的设计初衷。