缓存机制与记忆化（Memoization）

import functools import time # 未使用缓存的版本 def fib_no_cache(n): if n < 2: return n return fib_no_cache(n - 1) + fib_no_cache(n - 2) # 使用 lru_cache 的版本 @functools.lru_cache(maxsize=128) def fib_with_cache(n): if n < 2: return n return fib_with_cache(n - 1) + fib_with_cache(n - 2) # 性能对比 start = time.perf_counter() fib_no_cache(35) print(f"无缓存: {time.perf_counter() - start:.4f}s") start = time.perf_counter() fib_with_cache(35) print(f"有缓存: {time.perf_counter() - start:.4f}s") # 输出示例: # 无缓存: 2.3456s # 有缓存: 0.0001s

# 限制缓存大小 @functools.lru_cache(maxsize=32) def expensive_func(n): # 模拟耗时计算 time.sleep(0.1) return n ** 2 # maxsize 设置为 None 表示无限制 @functools.lru_cache(maxsize=None) def unlimited_cache(n): return n ** 2

@functools.lru_cache(typed=True) def process_value(x): return f"处理: {x} (类型: {type(x).__name__})" # typed=True 时，3 和 3.0 使用不同缓存条目 print(process_value(3)) # 计算并缓存 int 3 print(process_value(3.0)) # 计算并缓存 float 3.0（新条目） print(process_value(3)) # 命中 int 3 的缓存 # 输出: # 处理: 3 (类型: int) # 处理: 3.0 (类型: float) # 处理: 3 (类型: int)

@functools.lru_cache(maxsize=32) def compute(x): return x * x # 执行一系列调用 for i in range(10): compute(i) for i in range(5): compute(i) # 重复调用前5个值 info = compute.cache_info() print(f"命中: {info.hits}") print(f"未命中: {info.misses}") print(f"当前大小: {info.currsize}") print(f"最大容量: {info.maxsize}") print(f"命中率: {info.hits / (info.hits + info.misses):.2%}") # 输出: # 命中: 5 # 未命中: 10 # 当前大小: 10 # 最大容量: 32 # 命中率: 33.33%

@functools.lru_cache(maxsize=None) def get_user_data(user_id): print(f"从数据库获取用户 {user_id} 的数据...") return {"id": user_id, "name": f"User_{user_id}"} # 首次调用 - 实际计算 print(get_user_data(1)) # 输出: 从数据库获取用户 1 的数据... # 第二次调用 - 命中缓存 print(get_user_data(1)) # 不会打印"从数据库获取..." # 清空所有缓存 get_user_data.cache_clear() print(f"缓存已清空，当前大小: {get_user_data.cache_info().currsize}") # 再次调用 - 重新计算 print(get_user_data(1)) # 再次输出: 从数据库获取用户 1 的数据...

import functools import time def timed_lru_cache(seconds: int, maxsize: int = 128): """带 TTL 的 lru_cache 封装""" def wrapper(func): func = functools.lru_cache(maxsize=maxsize)(func) func.expire_time = seconds func.last_clear = time.time() @functools.wraps(func) def wrapped(*args, **kwargs): if time.time() - func.last_clear > func.expire_time: func.cache_clear() func.last_clear = time.time() return func(*args, **kwargs) return wrapped return wrapper @timed_lru_cache(seconds=5) def get_time_based_data(key): print(f"计算并缓存: {key}") return f"{key}_at_{time.time()}" print(get_time_based_data("a")) # 计算 print(get_time_based_data("a")) # 命中缓存 time.sleep(6) # 等待缓存过期 print(get_time_based_data("a")) # 缓存已过期，重新计算

import functools # Python 3.9+ 新语法，等价于 maxsize=None 的 lru_cache @functools.cache def load_config(path): print(f"加载配置文件: {path}") with open(path, 'r') as f: return f.read() # 首次加载 config = load_config("/etc/app/config.yaml") # 再次加载 - 直接从缓存返回 config = load_config("/etc/app/config.yaml") # 不会重新读取文件

import functools import time class DataAnalyzer: def __init__(self, data): self.data = data @property def regular_property(self): """每次访问都重新计算""" print("计算 regular_property...") time.sleep(1) return sum(self.data) / len(self.data) @functools.cached_property def cached_average(self): """只在首次访问时计算，后续直接返回缓存值""" print("计算 cached_average（仅一次）...") time.sleep(1) return sum(self.data) / len(self.data) analyzer = DataAnalyzer([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) # property - 每次重新计算 start = time.time() print(f"第1次: {analyzer.regular_property} ({time.time()-start:.2f}s)") start = time.time() print(f"第2次: {analyzer.regular_property} ({time.time()-start:.2f}s)") # cached_property - 仅首次计算 start = time.time() print(f"缓存第1次: {analyzer.cached_average} ({time.time()-start:.2f}s)") start = time.time() print(f"缓存第2次: {analyzer.cached_average} ({time.time()-start:.2f}s)") # 手动清空缓存属性 del analyzer.__dict__['cached_average'] print("缓存已清空") # 再次访问会重新计算 print(f"重新计算: {analyzer.cached_average}")

import functools import time def memoize(func): """简单的字典缓存记忆化装饰器""" cache = {} @functools.wraps(func) def wrapper(*args, **kwargs): # 使用参数元组作为缓存键 key = (args, tuple(sorted(kwargs.items()))) if key not in cache: cache[key] = func(*args, **kwargs) return cache[key] # 暴露缓存操作方法 wrapper.cache = cache wrapper.cache_info = lambda: { 'size': len(cache), 'hits': getattr(wrapper, '_hits', 0) } wrapper.cache_clear = cache.clear return wrapper @memoize def slow_add(a, b): time.sleep(0.5) return a + b print(slow_add(1, 2)) # 耗时 0.5s print(slow_add(1, 2)) # 立即返回（缓存命中） print(f"缓存大小: {len(slow_add.cache)}")

import functools from collections import OrderedDict def lru_cache_manual(maxsize=128): """手工实现的 LRU 缓存装饰器""" def decorator(func): cache = OrderedDict() hits = 0 misses = 0 @functools.wraps(func) def wrapper(*args, **kwargs): nonlocal hits, misses key = (args, tuple(sorted(kwargs.items()))) if key in cache: hits += 1 # LRU: 将访问的条目移到末尾 cache.move_to_end(key) return cache[key] misses += 1 result = func(*args, **kwargs) if len(cache) >= maxsize: # 移除最久未使用（最前面）的条目 cache.popitem(last=False) cache[key] = result return result wrapper.cache_info = lambda: { 'hits': hits, 'misses': misses, 'currsize': len(cache), 'maxsize': maxsize, 'hit_rate': f"{hits / (hits + misses) * 100:.1f}%" if (hits + misses) > 0 else "0%" } wrapper.cache_clear = lambda: cache.clear() return wrapper return decorator @lru_cache_manual(maxsize=3) def compute(x): print(f"计算: {x}") return x ** 2 compute(1) # 计算 compute(2) # 计算 compute(3) # 计算 compute(1) # 命中缓存，同时将 1 移到末尾 compute(4) # 计算，淘汰最久未使用的 2 compute(2) # 重新计算，因为 2 已被淘汰 print(compute.cache_info())

import functools import time def ttl_cache_manual(seconds=60, maxsize=128): """带 TTL（生存时间）的缓存装饰器""" def decorator(func): cache = {} # key -> (value, expiry_time) hits = 0 misses = 0 evictions = 0 @functools.wraps(func) def wrapper(*args, **kwargs): nonlocal hits, misses, evictions key = (args, tuple(sorted(kwargs.items()))) now = time.time() if key in cache: value, expiry = cache[key] if now < expiry: hits += 1 return value else: # 条目已过期 del cache[key] evictions += 1 misses += 1 result = func(*args, **kwargs) if len(cache) >= maxsize: # 淘汰最早过期的条目 oldest_key = min(cache, key=lambda k: cache[k][1]) del cache[oldest_key] evictions += 1 cache[key] = (result, now + seconds) return result wrapper.cache_info = lambda: { 'hits': hits, 'misses': misses, 'currsize': len(cache), 'maxsize': maxsize, 'evictions': evictions, 'ttl': seconds } wrapper.cache_clear = lambda: cache.clear() return wrapper return decorator @ttl_cache_manual(seconds=3) def fetch_data(url): print(f"实际请求: {url}") return f"" print(fetch_data("/api/users")) # 实际请求 print(fetch_data("/api/users")) # 缓存命中 time.sleep(4) # 等待 TTL 过期 print(fetch_data("/api/users")) # TTL过期，重新请求

# 安装 # pip install cachetools from cachetools import cached, LRUCache, TTLCache import time # LRU 缓存装饰器（等价于 functools.lru_cache） @cached(cache=LRUCache(maxsize=32)) def expensive_op(n): print(f"计算: {n}") time.sleep(0.5) return n * 2 # TTL 缓存装饰器（自动过期） @cached(cache=TTLCache(maxsize=100, ttl=30)) def get_weather(city): print(f"获取 {city} 天气...") return {"city": city, "temp": 25, "humidity": 60} print(get_weather("北京")) # 实际调用 print(get_weather("北京")) # 缓存命中 time.sleep(31) # 等待 TTL 过期 print(get_weather("北京")) # TTL过期，重新调用

from cachetools import cached, LRUCache, TTLCache from cachetools.keys import hashkey import time # 自定义缓存键（忽略某些参数） @cached( cache=TTLCache(maxsize=50, ttl=60), key=lambda url, timeout=30: hashkey(url) # 忽略 timeout 参数 ) def fetch_url(url, timeout=30): print(f"请求: {url}") time.sleep(0.2) return f"response from {url}" # 即使 timeout 不同，只要 url 相同就命中缓存 print(fetch_url("/api/data", timeout=30)) print(fetch_url("/api/data", timeout=60)) # 命中缓存！

from cachetools import cached, LFUCache # LFU（Least Frequently Used）缓存 # 淘汰访问频率最低的条目，适合热点数据场景 @cached(cache=LFUCache(maxsize=3)) def query_database(sql): print(f"执行SQL: {sql}") return f"result of {sql}" # 模拟访问模式：频繁访问某些SQL for _ in range(10): query_database("SELECT * FROM users") # 高频 for _ in range(5): query_database("SELECT * FROM orders") # 中频 query_database("SELECT * FROM products") # 低频 query_database("SELECT * FROM logs") # 新查询，淘汰低频的 products # 此时 products 已被淘汰，logs 被缓存

# 安装: pip install redis import redis import json import time # 连接 Redis r = redis.Redis( host='localhost', port=6379, db=0, decode_responses=True # 自动解码为字符串 ) # 基本缓存操作 cache_key = "user:1001" user_data = {"name": "张三", "age": 28, "city": "北京"} # 设置缓存，过期时间 300 秒 r.setex(cache_key, 300, json.dumps(user_data)) # 获取缓存 cached = r.get(cache_key) if cached: print(f"缓存命中: {json.loads(cached)}") else: print("缓存未命中") # 检查键是否存在 print(f"键存在: {r.exists(cache_key)}") # 获取剩余 TTL print(f"剩余时间: {r.ttl(cache_key)}秒")

import redis import json import functools import hashlib class RedisCache: """Redis 缓存装饰器""" def __init__(self, host='localhost', port=6379, db=0, default_ttl=300): self.client = redis.Redis( host=host, port=port, db=db, decode_responses=True ) self.default_ttl = default_ttl def __call__(self, ttl=None): ttl = ttl or self.default_ttl def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): # 生成缓存键 key_data = f"{func.__module__}:{func.__name__}:{args}:{kwargs}" cache_key = f"cache:{hashlib.md5(key_data.encode()).hexdigest()}" # 尝试从缓存获取 cached = self.client.get(cache_key) if cached is not None: return json.loads(cached) # 执行函数并缓存结果 result = func(*args, **kwargs) self.client.setex(cache_key, ttl, json.dumps(result)) return result wrapper.cache_clear = lambda: None # Redis缓存清理需单独实现 return wrapper return decorator # 使用示例 cache = RedisCache(default_ttl=600) @cache(ttl=300) # 单独设置5分钟过期 def get_user_info(user_id): print(f"从数据库查询用户 {user_id}") return {"id": user_id, "name": f"User_{user_id}", "timestamp": time.time()} # 首次调用 - 查询数据库并写入Redis info1 = get_user_info(1001) # 第二次调用 - 直接从Redis返回 info2 = get_user_info(1001) print(f"两次结果相同: {info1 == info2}")

# 缓存穿透防护：布隆过滤器 # 使用 RedisBloom 模块 try: r.bf().create('user_bloom', 0.01, 1000000) r.bf().add('user_bloom', 'user:1001') def safe_get_user(user_id): cache_key = f"user:{user_id}" # 布隆过滤器快速判断 if not r.bf().exists('user_bloom', cache_key): return None # 肯定不存在，直接返回 # 从缓存获取 result = r.get(cache_key) if result: return json.loads(result) # 从数据库查询（概率极小） return query_database(user_id) except redis.exceptions.ModuleError: print("RedisBloom 模块未安装") # 缓存击穿防护：互斥锁 def get_data_with_lock(key, compute_func, ttl=300): """使用互斥锁防止缓存击穿""" data = r.get(key) if data is not None: return json.loads(data) # 尝试获取锁 lock_key = f"lock:{key}" if r.setnx(lock_key, "1"): # 获取锁成功，计算数据 r.expire(lock_key, 10) # 防止死锁 data = compute_func() r.setex(key, ttl, json.dumps(data)) r.delete(lock_key) return data else: # 获取锁失败，等待后重试 time.sleep(0.1) return get_data_with_lock(key, compute_func, ttl)

from cachetools import LRUCache, LFUCache, FIFOCache, RRCache import time class CacheBenchmark: """缓存策略基准测试""" def __init__(self, cache, name): self.cache = cache self.name = name self.hits = 0 self.misses = 0 def access(self, key): if key in self.cache: self.hits += 1 return self.cache[key] else: self.misses += 1 # 模拟获取数据 result = f"data_{key}" self.cache[key] = result return result def report(self): total = self.hits + self.misses rate = self.hits / total * 100 if total > 0 else 0 return f"{self.name:10s} | 命中: {self.hits:4d} | 未命中: {self.misses:4d} | 命中率: {rate:5.1f}%" # 模拟80/20访问模式（80%的访问集中在20%的数据上） def simulate_8020(cache_obj, n_accesses=1000): for i in range(n_accesses): # 80%概率访问前20%的热点数据 if i % 5 != 0: key = f"key_{i % 20}" # 热点数据 else: key = f"key_{i % 100}" # 冷数据 cache_obj.access(key) # 运行测试 strategies = [ (LRUCache(maxsize=50), "LRU"), (LFUCache(maxsize=50), "LFU"), (FIFOCache(maxsize=50), "FIFO"), (RRCache(maxsize=50), "RR"), ] for cache_impl, name in strategies: bench = CacheBenchmark(cache_impl, name) simulate_8020(bench) print(bench.report()) # 典型输出: # LRU | 命中: 780 | 未命中: 220 | 命中率: 78.0% # LFU | 命中: 802 | 未命中: 198 | 命中率: 80.2% # FIFO | 命中: 720 | 未命中: 280 | 命中率: 72.0% # RR | 命中: 650 | 未命中: 350 | 命中率: 65.0%

import functools import time import random class CacheManager: """缓存管理器，支持多种失效策略""" def __init__(self): self._cache = {} self._version = 0 # -------- 策略1: Write-Through（穿透写入）-------- def write_through_set(self, key, value, db_func): """同时写入缓存和数据库""" self._cache[key] = value db_func(key, value) # 同步写数据库 print(f"[Write-Through] 缓存和数据库同步写入: {key}") # -------- 策略2: Write-Behind（异步写入）-------- def write_behind_set(self, key, value, db_func, delay=1): """先写入缓存，异步写数据库""" self._cache[key] = value print(f"[Write-Behind] 缓存已更新: {key}，数据库将在{delay}秒后更新") # 模拟异步任务 time.sleep(delay) db_func(key, value) # -------- 策略3: Cache-Aside（旁路缓存）-------- def cache_aside_get(self, key, db_func, ttl=60): """先查缓存，未命中则查数据库并更新缓存""" if key in self._cache: value, expiry = self._cache[key] if time.time() < expiry: print(f"[Cache-Aside] 缓存命中: {key}") return value else: # 懒淘汰 del self._cache[key] print(f"[Cache-Aside] 缓存未命中，查询数据库: {key}") value = db_func(key) self._cache[key] = (value, time.time() + ttl) return value # -------- 策略4: 主动失效（Invalidation）-------- def invalidate(self, key): """主动使缓存失效""" if key in self._cache: del self._cache[key] print(f"[Invalidation] 缓存已失效: {key}") def invalidate_pattern(self, pattern): """根据模式批量失效""" keys = [k for k in self._cache if pattern in k] for key in keys: del self._cache[key] print(f"[Invalidation] 批量失效 {len(keys)} 个缓存: pattern={pattern}") # -------- 策略5: 版本号控制-------- def versioned_get(self, key, version_func): """使用版本号确保缓存一致性""" current_version = version_func() if key in self._cache: cached_value, cached_version = self._cache[key] if cached_version == current_version: return cached_value else: print(f"[版本控制] 缓存版本过期: {cached_version} -> {current_version}") return None def versioned_set(self, key, value, version): self._cache[key] = (value, version) # 使用示例 cm = CacheManager() def save_to_db(key, value): print(f" 数据库已保存: {key}={value}") def load_from_db(key): print(f" 从数据库加载: {key}") return f"db_value_{key}" # Write-Through 测试 cm.write_through_set("user:1", "Alice", save_to_db) # Cache-Aside 测试 cm.invalidate("user:1") # 先失效 print(cm.cache_aside_get("user:1", load_from_db)) print(cm.cache_aside_get("user:1", load_from_db)) # 命中缓存

# 缓存穿透问题示意图 # 请求 -> 缓存未命中 -> 数据库查询 -> 未找到 -> 返回空 # 下次请求再次走完整流程... # 解决方案1：缓存空值 import functools def cache_null_solution(get_data_func, cache, ttl=60): """缓存空值方案：对不存在的结果也缓存""" def get_data(key): # 先查缓存 result = cache.get(key) if result is not None: return result if result != "__NULL__" else None # 查数据库 data = get_data_func(key) if data is None: # 对空结果也缓存，但TTL较短 cache.setex(key, ttl, "__NULL__") else: cache.set(key, data) return data return get_data # 解决方案2：布隆过滤器（Bloom Filter） # 在缓存层之前加一道屏障 class BloomFilter: """简单的布隆过滤器实现""" def __init__(self, size=100000, hash_count=3): self.size = size self.hash_count = hash_count self.bit_array = 0 def _hashes(self, item): """生成多个哈希值""" result = [] for i in range(self.hash_count): h = hash(f"{i}_{item}") result.append(h % self.size) return result def add(self, item): for pos in self._hashes(item): self.bit_array |= (1 << pos) def might_contain(self, item): for pos in self._hashes(item): if not (self.bit_array & (1 << pos)): return False return True # 使用布隆过滤器 bf = BloomFilter() # 预热：将所有存在的ID加入布隆过滤器 for i in range(10000): bf.add(f"user_{i}") def safe_query_user(user_id): key = f"user_{user_id}" if not bf.might_contain(key): return None # 确定不存在，直接返回 # 继续查缓存和数据库... return query_database(user_id)

# 缓存雪崩问题 # 解决方案1：缓存过期时间增加随机偏移 import random def set_cache_with_jitter(key, value, base_ttl=300): """设置缓存，过期时间加入随机偏移防止同时过期""" jitter = random.uniform(0, 60) # 0~60秒随机偏移 ttl = base_ttl + jitter cache.setex(key, int(ttl), value) # 批量初始化时使用 def init_cache_batch(data_dict, base_ttl=300): for key, value in data_dict.items(): set_cache_with_jitter(key, value, base_ttl) # 解决方案2：多级缓存 def multi_level_get(key, l1_cache, l2_cache, db_func, l1_ttl=60, l2_ttl=600): """ 多级缓存： L1（本地内存）-> L2（Redis）-> 数据库 L1的TTL短，L2的TTL长 """ # L1 缓存 result = l1_cache.get(key) if result: return result # L2 缓存（Redis） result = l2_cache.get(key) if result: # 异步回填L1（使用较短TTL） l1_cache.setex(key, l1_ttl, result) return result # 数据库 result = db_func(key) if result: l2_cache.setex(key, l2_ttl, result) l1_cache.setex(key, l1_ttl, result) return result # 解决方案3：限流与熔断（以Redis实现） import time class RateLimiter: """滑动窗口限流器""" def __init__(self, redis_client, max_requests=100, window=1): self.redis = redis_client self.max_requests = max_requests self.window = window def allow_request(self, key="global"): now = time.time() window_key = f"ratelimit:{key}" # 移除窗口外的记录 self.redis.zremrangebyscore(window_key, 0, now - self.window) # 检查请求数 count = self.redis.zcard(window_key) if count >= self.max_requests: return False # 添加当前请求 self.redis.zadd(window_key, {f"{now}_{random.random()}": now}) self.redis.expire(window_key, self.window + 1) return True # 使用限流保护数据库 limiter = RateLimiter(redis_client, max_requests=10, window=1) def protected_query(db_func, key): if not limiter.allow_request("db_query"): raise Exception("请求过于频繁，请稍后重试") return db_func(key)

# 缓存击穿问题：热点key过期瞬间的高并发 # 解决方案1：互斥锁（最常用） import threading class HotKeyProtector: """热点数据保护器（互斥锁方案）""" def __init__(self): self._locks = {} self._lock_lock = threading.Lock() def get_or_compute(self, key, compute_func, cache_get, cache_set, ttl=300): """ 使用互斥锁保护热点key 只允许一个线程/进程查询数据库，其他等待 """ # 先尝试直接从缓存获取 result = cache_get(key) if result is not None: return result # 获取或创建该key的锁 with self._lock_lock: if key not in self._locks: self._locks[key] = threading.Lock() key_lock = self._locks[key] # 加锁（只允许一个线程通过） with key_lock: # 双重检查：可能等待锁期间其他线程已经填充了缓存 result = cache_get(key) if result is not None: return result # 查询数据库 result = compute_func() cache_set(key, result, ttl) return result # 解决方案2：逻辑过期（不设置实际过期时间） import dataclasses import time @dataclasses.dataclass class CacheItem: """带逻辑过期时间的缓存条目""" value: any expire_time: float class LogicalExpiryCache: """逻辑过期缓存：数据永不过期，但标记逻辑过期""" def __init__(self, compute_func, ttl=300): self._cache = {} self._ttl = ttl self._compute = compute_func self._lock = threading.Lock() def get(self, key): now = time.time() if key in self._cache: item = self._cache[key] if now < item.expire_time: return item.value # 未过期，直接返回 else: # 逻辑过期：异步刷新 self._async_refresh(key) # 返回旧数据（牺牲一致性换可用性） return item.value # 缓存不存在：同步加载 return self._sync_load(key) def _async_refresh(self, key): """异步刷新缓存""" def refresh(): new_value = self._compute(key) self._cache[key] = CacheItem( value=new_value, expire_time=time.time() + self._ttl ) thread = threading.Thread(target=refresh, daemon=True) thread.start() def _sync_load(self, key): """同步加载缓存""" with self._lock: # 双重检查 if key in self._cache: item = self._cache[key] if time.time() < item.expire_time: return item.value value = self._compute(key) self._cache[key] = CacheItem( value=value, expire_time=time.time() + self._ttl ) return value

import functools import time import threading import json import logging from collections import OrderedDict from dataclasses import dataclass, field from typing import Any, Callable, Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class CacheStats: """缓存统计信息""" hits: int = 0 misses: int = 0 evictions: int = 0 total_latency: float = 0.0 @property def hit_rate(self) -> float: total = self.hits + self.misses return self.hits / total if total > 0 else 0.0 @property def avg_latency(self) -> float: total = self.hits + self.misses return self.total_latency / total if total > 0 else 0.0 class TieredCache: """ 多级缓存系统 L1: 本地内存（LRU，最快） L2: 本地TTL缓存 L3: 外部Redis（可选） """ def __init__(self, l1_maxsize: int = 128, l2_ttl: int = 60, l2_maxsize: int = 1024, enable_redis: bool = False, redis_client=None): self.l1 = OrderedDict() # LRU缓存 self.l1_maxsize = l1_maxsize self.l2 = {} # TTL缓存 self.l2_ttl = l2_ttl self.l2_maxsize = l2_maxsize self.redis = redis_client if enable_redis else None self.stats = CacheStats() self._lock = threading.Lock() def get(self, key: str, compute_func: Callable, ttl: Optional[int] = None) -> Any: """ 多级缓存查询 顺序: L1 -> L2 -> Redis -> compute_func """ start = time.perf_counter() ttl = ttl or self.l2_ttl # L1 缓存（最快的内存LRU） result = self._l1_get(key) if result is not None: self.stats.hits += 1 self.stats.total_latency += time.perf_counter() - start return result # L2 缓存（TTL内存缓存） result = self._l2_get(key) if result is not None: self._l1_set(key, result) self.stats.hits += 1 self.stats.total_latency += time.perf_counter() - start return result # Redis 缓存 if self.redis: result = self.redis.get(key) if result is not None: result = json.loads(result) self._l1_set(key, result) self._l2_set(key, result, ttl) self.stats.hits += 1 self.stats.total_latency += time.perf_counter() - start return result # 全部未命中，执行计算函数 self.stats.misses += 1 result = compute_func() # 回填各级缓存 self._l1_set(key, result) self._l2_set(key, result, ttl) if self.redis: self.redis.setex(key, ttl, json.dumps(result)) self.stats.total_latency += time.perf_counter() - start return result def invalidate(self, key: str): """使所有级别的缓存失效""" self.l1.pop(key, None) self.l2.pop(key, None) if self.redis: self.redis.delete(key) logger.info(f"缓存已失效: {key}") def _l1_get(self, key: str) -> Optional[Any]: if key in self.l1: self.l1.move_to_end(key) # LRU: 移到末尾 return self.l1[key] return None def _l1_set(self, key: str, value: Any): if len(self.l1) >= self.l1_maxsize: self.l1.popitem(last=False) # 淘汰最久未使用 self.l1[key] = value def _l2_get(self, key: str) -> Optional[Any]: if key in self.l2: value, expiry = self.l2[key] if time.time() < expiry: return value else: del self.l2[key] self.stats.evictions += 1 return None def _l2_set(self, key: str, value: Any, ttl: int): if len(self.l2) >= self.l2_maxsize: # 淘汰最早过期的 oldest = min(self.l2, key=lambda k: self.l2[k][1]) del self.l2[oldest] self.l2[key] = (value, time.time() + ttl) def stats_report(self) -> dict: return { "l1_size": len(self.l1), "l2_size": len(self.l2), "hits": self.stats.hits, "misses": self.stats.misses, "hit_rate": f"{self.stats.hit_rate:.1%}", "avg_latency_ms": f"{self.stats.avg_latency * 1000:.2f}ms", "evictions": self.stats.evictions, } # 使用示例 cache = TieredCache(l1_maxsize=100, l2_ttl=60) def expensive_query(query_id): time.sleep(0.5) # 模拟耗时 return {"id": query_id, "data": f"result_{query_id}", "time": time.time()} # 首次查询（全部未命中） result1 = cache.get("query:1", lambda: expensive_query(1)) print(f"首次查询: {result1}") # 二次查询（命中L1） result2 = cache.get("query:1", lambda: expensive_query(1)) print(f"再次查询: {result2}") # 查看统计 print(f"缓存统计: {cache.stats_report()}")