本文目录导读：
我来详细介绍Python实现多级缓存架构的方法。
基础多级缓存架构
import time
import json
from functools import wraps
from typing import Optional, Any, Callable
import redis
from collections import OrderedDict
class CacheLevel:
"""缓存层级基类"""
def __init__(self, name: str, ttl: int = 300):
self.name = name
self.ttl = ttl
def get(self, key: str) -> Optional[Any]:
raise NotImplementedError
def set(self, key: str, value: Any, ttl: Optional[int] = None):
raise NotImplementedError
def delete(self, key: str):
raise NotImplementedError
def clear(self):
raise NotImplementedError
class LocalCache(CacheLevel):
"""本地内存缓存（L1）"""
def __init__(self, name: str = "L1", ttl: int = 60, maxsize: int = 1000):
super().__init__(name, ttl)
self.cache = OrderedDict()
self.maxsize = maxsize
def get(self, key: str) -> Optional[Any]:
if key in self.cache:
value, expiry = self.cache[key]
if expiry is None or expiry > time.time():
# LRU: 移动到末尾
self.cache.move_to_end(key)
return value
else:
del self.cache[key]
return None
def set(self, key: str, value: Any, ttl: Optional[int] = None):
ttl = ttl or self.ttl
expiry = time.time() + ttl if ttl else None
# LRU淘汰
if len(self.cache) >= self.maxsize:
self.cache.popitem(last=False)
self.cache[key] = (value, expiry)
def delete(self, key: str):
self.cache.pop(key, None)
def clear(self):
self.cache.clear()
class RedisCache(CacheLevel):
"""Redis缓存（L2）"""
def __init__(self, name: str = "L2", ttl: int = 600,
host: str = 'localhost', port: int = 6379, db: int = 0):
super().__init__(name, ttl)
self.client = redis.Redis(host=host, port=port, db=db,
decode_responses=True)
def get(self, key: str) -> Optional[Any]:
value = self.client.get(key)
if value:
try:
return json.loads(value)
except:
return value
return None
def set(self, key: str, value: Any, ttl: Optional[int] = None):
ttl = ttl or self.ttl
if isinstance(value, (dict, list, tuple)):
value = json.dumps(value)
self.client.setex(key, ttl, value)
def delete(self, key: str):
self.client.delete(key)
def clear(self):
self.client.flushdb()
class MultiLevelCache:
"""多级缓存管理器"""
def __init__(self):
self.levels = []
def add_level(self, cache: CacheLevel):
"""添加缓存层级"""
self.levels.append(cache)
def get(self, key: str) -> Optional[Any]:
"""从各级缓存获取数据"""
for i, level in enumerate(self.levels):
value = level.get(key)
if value is not None:
# 如果从下级缓存找到，回填到上级缓存
if i > 0:
self._backfill(key, value, i-1)
return value
return None
def set(self, key: str, value: Any, ttl: Optional[int] = None):
"""设置所有缓存层级"""
for level in self.levels:
level.set(key, value, ttl=ttl)
def delete(self, key: str):
"""从所有缓存层级删除"""
for level in self.levels:
level.delete(key)
def clear(self):
"""清空所有缓存"""
for level in self.levels:
level.clear()
def _backfill(self, key: str, value: Any, up_to_level: int):
"""回填缓存"""
for i in range(up_to_level, -1, -1):
self.levels[i].set(key, value)
def get_or_compute(self, key: str, compute_func: Callable,
ttl: Optional[int] = None) -> Any:
"""获取或计算"""
value = self.get(key)
if value is None:
value = compute_func()
self.set(key, value, ttl=ttl)
return value
# 使用示例
cache = MultiLevelCache()
cache.add_level(LocalCache(ttl=60, maxsize=1000)) # L1: 本地缓存，60秒过期
cache.add_level(RedisCache(ttl=600)) # L2: Redis缓存，10分钟过期
装饰器实现
def multi_level_cache(cache_manager: MultiLevelCache, key_prefix: str = ""):
"""多级缓存装饰器"""
def decorator(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
# 生成缓存key
key = f"{key_prefix}:{func.__name__}:{str(args)}:{str(kwargs)}"
# 尝试从缓存获取
result = cache_manager.get(key)
if result is not None:
return result
# 计算并缓存
result = func(*args, **kwargs)
cache_manager.set(key, result)
return result
return wrapper
return decorator
# 使用示例
cache_manager = MultiLevelCache()
cache_manager.add_level(LocalCache())
cache_manager.add_level(RedisCache())
@multi_level_cache(cache_manager, key_prefix="user")
def get_user_info(user_id: int):
# 模拟数据库查询
time.sleep(1)
return {"id": user_id, "name": f"User_{user_id}"}
高级特性实现
1 缓存穿透保护
class BloomFilterCache(CacheLevel):
"""布隆过滤器缓存（防止缓存穿透）"""
def __init__(self, name: str = "Bloom", capacity: int = 1000000,
error_rate: float = 0.01):
super().__init__(name, ttl=0)
from bloom_filter import BloomFilter
self.bloom = BloomFilter(max_elements=capacity,
error_rate=error_rate)
self.cache = {}
def might_contain(self, key: str) -> bool:
return self.bloom.__contains__(key)
def add(self, key: str):
self.bloom.add(key)
# 增强的多级缓存
class AdvancedMultiLevelCache(MultiLevelCache):
def __init__(self):
super().__init__()
self.bloom_filter = BloomFilterCache()
def get(self, key: str) -> Optional[Any]:
# 布隆过滤器检查
if not self.bloom_filter.might_contain(key):
return None
return super().get(key)
def set(self, key: str, value: Any, ttl: Optional[int] = None):
# 添加布隆过滤器
self.bloom_filter.add(key)
super().set(key, value, ttl=ttl)
2 缓存雪崩保护
import random
class StaggeredExpiryCache(LocalCache):
"""带随机过期时间的缓存（防止缓存雪崩）"""
def set(self, key: str, value: Any, ttl: Optional[int] = None):
# 在基础TTL上增加随机时间
base_ttl = ttl or self.ttl
jitter = random.uniform(0, 0.2 * base_ttl) # 20%的抖动
actual_ttl = base_ttl + jitter
super().set(key, value, ttl=actual_ttl)
3 缓存热点保护
class HotKeyCache(LocalCache):
"""热key缓存（多级副本）"""
def __init__(self, name: str = "HotKey", ttl: int = 30,
hot_threshold: int = 100):
super().__init__(name, ttl)
self.access_count = {}
self.hot_threshold = hot_threshold
self.replicas = [{} for _ in range(3)] # 3个副本
def get(self, key: str) -> Optional[Any]:
# 更新访问计数
self.access_count[key] = self.access_count.get(key, 0) + 1
if self.access_count.get(key, 0) > self.hot_threshold:
# 热key：从多个副本随机读取
import random
replica_idx = random.randint(0, len(self.replicas) - 1)
return self.replicas[replica_idx].get(key)
return super().get(key)
完整应用示例
import asyncio
import logging
from typing import Optional
class CacheService:
"""缓存服务（生产级）"""
def __init__(self):
self.logger = logging.getLogger(__name__)
self.cache = MultiLevelCache()
# 配置缓存层级
self.cache.add_level(
LocalCache(ttl=60, maxsize=10000) # L1: 1分钟
)
self.cache.add_level(
RedisCache(ttl=600) # L2: 10分钟
)
async def get_data(self, key: str,
fetch_func: Callable,
ttl: Optional[int] = None) -> Any:
"""获取数据的完整流程"""
try:
# 尝试从缓存获取
data = self.cache.get(key)
if data:
self.logger.debug(f"Cache hit: {key}")
return data
# 缓存未命中，从数据源获取
self.logger.debug(f"Cache miss: {key}")
data = await fetch_func()
# 设置缓存
if data is not None:
self.cache.set(key, data, ttl=ttl)
return data
except Exception as e:
self.logger.error(f"Cache error: {e}")
# 降级：直接从数据源获取
return await fetch_func()
async def invalidate(self, key: str):
"""使缓存失效"""
self.cache.delete(key)
self.logger.info(f"Cache invalidated: {key}")
# 使用示例
async def main():
cache_service = CacheService()
# 模拟数据获取
async def fetch_user(user_id: int):
await asyncio.sleep(0.5) # 模拟数据库查询
return {"id": user_id, "name": f"User_{user_id}"}
# 获取数据（首次会从数据库加载）
user = await cache_service.get_data("user:1",
lambda: fetch_user(1))
print("First fetch:", user)
# 第二次获取（从缓存快速返回）
user = await cache_service.get_data("user:1",
lambda: fetch_user(1))
print("Second fetch:", user)
# 使缓存失效
await cache_service.invalidate("user:1")
# 运行
asyncio.run(main())
监控和统计
class CacheStats:
"""缓存统计"""
def __init__(self):
self.hits = 0
self.misses = 0
self.sets = 0
def hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0
class MonitoredCache(MultiLevelCache):
"""带监控的多级缓存"""
def __init__(self):
super().__init__()
self.stats = CacheStats()
def get(self, key: str) -> Optional[Any]:
value = super().get(key)
if value is not None:
self.stats.hits += 1
else:
self.stats.misses += 1
return value
def set(self, key: str, value: Any, ttl: Optional[int] = None):
self.stats.sets += 1
super().set(key, value, ttl=ttl)
这个多级缓存架构提供了：
- 分级缓存：L1（内存）-> L2（Redis）
- 缓存穿透保护：布隆过滤器
- 缓存雪崩保护：随机过期时间
- 热点缓存：多副本读取
- 监控统计：命中率统计
可以根据实际需求选择合适的缓存级别和配置参数。
