系统设计面试中的缓存策略从本地缓存到分布式一致性的考察要点一、面试中的缓存盲区只会 Redis 却答不出一致性代价系统设计面试中缓存几乎是必考话题。多数候选人能说出用 Redis 做缓存但追问缓存与数据库的一致性如何保证、本地缓存与分布式缓存如何选型、缓存穿透和雪崩的防护策略时回答往往停留在概念层面。某大厂面试统计显示系统设计环节中缓存相关问题的通过率仅 35%核心失分点在于无法量化不同一致性模型的性能代价也无法在具体业务场景下做出合理的缓存选型决策。缓存策略的工程本质是用空间换时间、用一致性换性能。理解各种缓存策略的适用边界和一致性代价是系统设计面试的区分度关键。二、缓存策略的分层架构与一致性模型flowchart TB subgraph 缓存分层[缓存分层架构] direction TB L1[L1 本地缓存br/进程内 HashMap/Caffeinebr/延迟 1msbr/容量有限] L2[L2 分布式缓存br/Redis/Memcachedbr/延迟 1-5msbr/容量可扩展] L3[L3 持久化存储br/MySQL/PostgreSQLbr/延迟 5-50msbr/数据可靠] end subgraph 一致性模型[一致性模型对比] direction LR SC[强一致性br/写穿透/写失效br/性能代价高] -- WC[弱一致性br/异步刷新br/延迟窗口] WC -- EC[最终一致性br/定时同步br/最大延迟窗口] end L1 --|缓存失效| L2 L2 --|缓存未命中| L3 一致性模型 --|选型依据| L1 一致性模型 --|选型依据| L2 style 缓存分层 fill:#eef,stroke:#333 style 一致性模型 fill:#efe,stroke:#333缓存一致性模型的选择取决于业务对读到旧数据的容忍度。金融账户余额要求强一致性社交动态的点赞数可以接受最终一致性。面试中需要根据业务场景明确选择哪种模型并解释其代价。三、缓存策略的代码实现与面试要点import time import threading from typing import Optional, Callable, Any from collections import OrderedDict from functools import wraps # 核心1本地缓存实现 class LRUCache: LRU 本地缓存基于 OrderedDict 实现 适用于单进程内的高频热点数据缓存 def __init__(self, capacity: int, ttl: Optional[float] None): self._capacity capacity self._ttl ttl self._cache: OrderedDict OrderedDict() self._timestamps: dict {} self._lock threading.Lock() self._hits 0 self._misses 0 def get(self, key: str) - Optional[Any]: with self._lock: if key not in self._cache: self._misses 1 return None # TTL 过期检查 if self._ttl and time.time() - self._timestamps[key] self._ttl: del self._cache[key] del self._timestamps[key] self._misses 1 return None # 移到末尾标记为最近使用 self._cache.move_to_end(key) self._hits 1 return self._cache[key] def put(self, key: str, value: Any): with self._lock: if key in self._cache: self._cache.move_to_end(key) else: if len(self._cache) self._capacity: # 淘汰最久未使用的 evicted_key next(iter(self._cache)) del self._cache[evicted_key] self._timestamps.pop(evicted_key, None) self._cache[key] value self._timestamps[key] time.time() def invalidate(self, key: str): 主动失效——写失效策略的核心操作 with self._lock: self._cache.pop(key, None) self._timestamps.pop(key, None) def hit_rate(self) - float: total self._hits self._misses return self._hits / total if total 0 else 0.0 # 核心2缓存策略模式 class CacheAside: Cache-Aside旁路缓存最常用的缓存策略 读先查缓存未命中则查数据库并回填 写先更新数据库再删除缓存 def __init__(self, cache: LRUCache, db_get: Callable, db_set: Callable): self._cache cache self._db_get db_get self._db_set db_set def get(self, key: str) - Optional[Any]: value self._cache.get(key) if value is not None: return value # 缓存未命中查询数据库 value self._db_get(key) if value is not None: self._cache.put(key, value) return value def set(self, key: str, value: Any): # 先更新数据库 self._db_set(key, value) # 再删除缓存而非更新缓存 self._cache.invalidate(key) class WriteThrough: Write-Through写穿透缓存与数据库同步写入 写延迟等于缓存写入 数据库写入 适用于写少读多且对一致性要求高的场景 def __init__(self, cache: LRUCache, db_get: Callable, db_set: Callable): self._cache cache self._db_get db_get self._db_set db_set def get(self, key: str) - Optional[Any]: value self._cache.get(key) if value is not None: return value value self._db_get(key) if value is not None: self._cache.put(key, value) return value def set(self, key: str, value: Any): # 同时写入缓存和数据库 self._cache.put(key, value) self._db_set(key, value) class WriteBehind: Write-Behind写回先写缓存异步批量写数据库 写延迟极低但存在数据丢失风险 适用于写密集且可容忍少量丢失的场景如计数器 def __init__(self, cache: LRUCache, db_batch_set: Callable, flush_interval: float 5.0): self._cache cache self._db_batch_set db_batch_set self._dirty_keys: set set() self._lock threading.Lock() self._flush_interval flush_interval self._running True # 启动异步刷盘线程 self._flush_thread threading.Thread(targetself._flush_loop, daemonTrue) self._flush_thread.start() def get(self, key: str) - Optional[Any]: return self._cache.get(key) def set(self, key: str, value: Any): with self._lock: self._cache.put(key, value) self._dirty_keys.add(key) def _flush_loop(self): 异步刷盘定期将脏数据写入数据库 while self._running: time.sleep(self._flush_interval) self._flush() def _flush(self): with self._lock: if not self._dirty_keys: return batch {} for key in list(self._dirty_keys): value self._cache.get(key) if value is not None: batch[key] value self._dirty_keys.clear() if batch: self._db_batch_set(batch) def shutdown(self): self._running False self._flush() # 最终刷盘 # 核心3缓存防护策略 class CacheProtection: 缓存三大问题的防护穿透、雪崩、击穿 def __init__(self, cache: LRUCache, db_get: Callable): self._cache cache self._db_get db_get # 空值缓存防止缓存穿透 self._null_cache: dict {} self._null_ttl 60 # 空值缓存 60 秒 # 互斥锁防止缓存击穿 self._mutex_locks: dict {} self._lock threading.Lock() def get_with_protection(self, key: str) - Optional[Any]: # 1. 查缓存 value self._cache.get(key) if value is not None: return value # 2. 查空值缓存防穿透 if key in self._null_cache: if time.time() - self._null_cache[key] self._null_ttl: return None # 数据库中确实不存在 del self._null_cache[key] # 3. 互斥锁防击穿 with self._lock: if key not in self._mutex_locks: self._mutex_locks[key] threading.Lock() mutex self._mutex_locks[key] with mutex: # 双重检查 value self._cache.get(key) if value is not None: return value # 查数据库 value self._db_get(key) if value is not None: self._cache.put(key, value) else: # 空值缓存防止穿透 self._null_cache[key] time.time() return value # 核心4多级缓存 class MultiLevelCache: 多级缓存L1 本地 L2 分布式 L1 命中率约 60%L2 命中率约 30%数据库承载约 10% def __init__(self, l1: LRUCache, l2_get: Callable, l2_set: Callable, db_get: Callable): self._l1 l1 self._l2_get l2_get self._l2_set l2_set self._db_get db_get def get(self, key: str) - Optional[Any]: # L1 本地缓存 value self._l1.get(key) if value is not None: return value # L2 分布式缓存 value self._l2_get(key) if value is not None: self._l1.put(key, value) # 回填 L1 return value # 数据库 value self._db_get(key) if value is not None: self._l1.put(key, value) self._l2_set(key, value) return value def invalidate(self, key: str): 多级缓存失效L1 L2 同时失效 self._l1.invalidate(key) # L2 失效通常通过 Redis DEL 命令 self._l2_set(key, None) # 实际应调用 Redis DELETE四、缓存策略的 Trade-offsCache-Aside 的双写不一致窗口。Cache-Aside 策略中先更新数据库、再删除缓存存在短暂的不一致窗口线程 A 更新数据库后、删除缓存前线程 B 读到旧缓存。解决方案包括延迟双删删除缓存后延迟再删一次和基于 binlog 的异步失效但前者引入延迟、后者增加系统复杂度。Write-Through 的写延迟。写穿透策略保证强一致性但每次写入都需等待数据库完成写延迟等于缓存写入 数据库写入。在高并发写入场景下数据库成为瓶颈。Write-Behind 的数据丢失风险。写回策略将写操作先缓存在内存中异步批量写入数据库。进程崩溃时未刷盘的脏数据会丢失。适用于计数器、浏览量等可容忍少量丢失的场景不适用于金融交易。本地缓存的集群一致性问题。多实例部署时每个实例的本地缓存独立一个实例更新数据后其他实例的本地缓存仍是旧值。解决方案包括短 TTL 自动过期、基于 Pub/Sub 的缓存失效通知、或放弃本地缓存只使用分布式缓存。每种方案都有各自的延迟和复杂度代价。五、总结缓存策略的选型核心是在一致性、延迟和复杂度之间做权衡。Cache-Aside 适合读多写少的通用场景Write-Through 适合一致性要求高的场景Write-Behind 适合写密集且可容忍丢失的场景。面试中需要根据业务特征读写比、一致性要求、数据量级做出选型决策并量化不同策略的代价。缓存防护穿透、雪崩、击穿是生产级系统的必备能力多级缓存是高并发场景下的常见架构模式。理解每种策略的适用边界和一致性代价远比记住用 Redis 做缓存更有区分度。