本文目录导读：
我来帮你实现一个完整的Python缓存工具封装,包含多种缓存策略和最佳实践。
基础缓存封装
1 简单内存缓存
import time
import threading
from functools import wraps
from typing import Any, Optional, Dict, Callable
class SimpleCache:
"""简单内存缓存"""
def __init__(self, default_ttl: int = 300):
self._cache: Dict[str, tuple] = {}
self._default_ttl = default_ttl
self._lock = threading.RLock()
def get(self, key: str) -> Optional[Any]:
"""获取缓存值"""
with self._lock:
if key not in self._cache:
return None
value, expiry = self._cache[key]
if expiry and time.time() > expiry:
del self._cache[key]
return None
return value
def set(self, key: str, value: Any, ttl: Optional[int] = None):
"""设置缓存"""
with self._lock:
expiry = time.time() + (ttl or self._default_ttl)
self._cache[key] = (value, expiry)
def delete(self, key: str):
"""删除缓存"""
with self._lock:
self._cache.pop(key, None)
def clear(self):
"""清空缓存"""
with self._lock:
self._cache.clear()
def __contains__(self, key: str) -> bool:
return self.get(key) is not None
2 装饰器实现
class cached:
"""缓存装饰器"""
def __init__(self, ttl: int = 300, cache: Optional[SimpleCache] = None):
self.ttl = ttl
self.cache = cache or SimpleCache()
def __call__(self, func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
# 创建缓存键
key = self._make_key(func.__name__, args, kwargs)
# 尝试获取缓存
result = self.cache.get(key)
if result is not None:
return result
# 执行函数
result = func(*args, **kwargs)
self.cache.set(key, result, self.ttl)
return result
return wrapper
@staticmethod
def _make_key(func_name: str, args: tuple, kwargs: dict) -> str:
"""生成缓存键"""
key_parts = [func_name]
key_parts.extend(str(arg) for arg in args)
key_parts.extend(f"{k}:{v}" for k, v in sorted(kwargs.items()))
return ":".join(key_parts)
高级缓存封装
1 分层缓存实现
import json
import pickle
from typing import List, Optional, Callable
class RedisCache:
"""Redis缓存适配器"""
def __init__(self, redis_client, prefix: str = "cache:"):
self.client = redis_client
self.prefix = prefix
def get(self, key: str) -> Optional[Any]:
value = self.client.get(f"{self.prefix}{key}")
if value:
return pickle.loads(value)
return None
def set(self, key: str, value: Any, ttl: int = 300):
self.client.setex(
f"{self.prefix}{key}",
ttl,
pickle.dumps(value)
)
def delete(self, key: str):
self.client.delete(f"{self.prefix}{key}")
class MultiLevelCache:
"""多级缓存"""
def __init__(self, caches: List[object]):
self.caches = caches
def get(self, key: str) -> Optional[Any]:
for i, cache in enumerate(self.caches):
value = cache.get(key)
if value is not None:
# 回写上一级缓存
for j in range(i - 1, -1, -1):
self.caches[j].set(key, value)
return value
return None
def set(self, key: str, value: Any, ttl: int = 300):
for cache in self.caches:
cache.set(key, value, ttl)
def delete(self, key: str):
for cache in self.caches:
cache.delete(key)
# 使用示例
# memory_cache = SimpleCache(ttl=60)
# redis_cache = RedisCache(redis_client)
# cache = MultiLevelCache([memory_cache, redis_cache])
2 缓存管理器封装
from enum import Enum
from typing import Dict, Any, Optional
import hashlib
import logging
logger = logging.getLogger(__name__)
class CacheStrategy(Enum):
"""缓存策略"""
LRU = "lru"
LFU = "lfu"
FIFO = "fifo"
TTL = "ttl"
class CacheManager:
"""缓存管理器"""
def __init__(self,
default_ttl: int = 300,
max_size: int = 1000,
strategy: CacheStrategy = CacheStrategy.TTL):
self.default_ttl = default_ttl
self.max_size = max_size
self.strategy = strategy
self._cache: Dict[str, Any] = {}
self._expiry: Dict[str, float] = {}
self._access_count: Dict[str, int] = {}
self._access_order: list = []
def get(self, key: str) -> Optional[Any]:
"""获取缓存"""
if key not in self._cache:
return None
# 检查过期
if self._is_expired(key):
self.delete(key)
return None
# 更新访问信息
self._update_access(key)
logger.debug(f"Cache hit: {key}")
return self._cache[key]
def set(self, key: str, value: Any, ttl: Optional[int] = None):
"""设置缓存"""
# 检查容量
if len(self._cache) >= self.max_size and key not in self._cache:
self._evict_one()
# 设置缓存
self._cache[key] = value
self._expiry[key] = time.time() + (ttl or self.default_ttl)
self._update_access(key)
logger.debug(f"Cache set: {key}")
def delete(self, key: str):
"""删除缓存"""
self._cache.pop(key, None)
self._expiry.pop(key, None)
self._access_count.pop(key, None)
if key in self._access_order:
self._access_order.remove(key)
def clear(self):
"""清空缓存"""
self._cache.clear()
self._expiry.clear()
self._access_count.clear()
self._access_order.clear()
def _is_expired(self, key: str) -> bool:
"""检查是否过期"""
expiry = self._expiry.get(key)
return expiry and time.time() > expiry
def _update_access(self, key: str):
"""更新访问信息"""
# 更新访问次数
self._access_count[key] = self._access_count.get(key, 0) + 1
# 更新访问顺序
if key in self._access_order:
self._access_order.remove(key)
self._access_order.append(key)
def _evict_one(self):
"""淘汰一个缓存项"""
if not self._cache:
return
if self.strategy == CacheStrategy.LRU:
# 淘汰最久未使用的
key = self._access_order[0]
elif self.strategy == CacheStrategy.LFU:
# 淘汰使用频率最低的
key = min(self._access_count, key=self._access_count.get)
else:
# 默认淘汰第一个
key = next(iter(self._cache))
self.delete(key)
logger.debug(f"Cache evicted: {key}")
def get_stats(self) -> dict:
"""获取缓存统计信息"""
return {
'size': len(self._cache),
'max_size': self.max_size,
'strategy': self.strategy.value,
'access_count': sum(self._access_count.values())
}
# 工厂函数
def create_cache(backend: str = 'memory', **kwargs) -> Any:
"""创建缓存实例"""
backends = {
'memory': SimpleCache,
'manager': CacheManager,
'redis': RedisCache
}
cache_class = backends.get(backend)
if not cache_class:
raise ValueError(f"Unknown cache backend: {backend}")
return cache_class(**kwargs)
实际应用示例
1 数据库查询缓存
class DatabaseCache:
"""数据库查询缓存"""
def __init__(self, cache_instance):
self.cache = cache_instance
def query_with_cache(self, query: str, ttl: int = 60):
"""带缓存的数据库查询"""
cache_key = self._generate_key(query)
# 尝试获取缓存
result = self.cache.get(cache_key)
if result is not None:
return result
# 执行查询
result = self._execute_query(query)
# 缓存结果
self.cache.set(cache_key, result, ttl)
return result
@staticmethod
def _generate_key(query: str) -> str:
"""生成缓存键"""
return f"db:{hashlib.md5(query.encode()).hexdigest()}"
def _execute_query(self, query: str):
"""执行数据库查询"""
# 实际数据库查询逻辑
pass
2 API响应缓存
class APICache:
"""API响应缓存"""
def __init__(self, cache_instance):
self.cache = cache_instance
def cached_api_request(self,
url: str,
method: str = 'GET',
headers: dict = None,
params: dict = None,
ttl: int = 300):
"""带缓存的API请求"""
cache_key = self._make_cache_key(url, method, params)
# 检查缓存
if method.upper() == 'GET':
cached_response = self.cache.get(cache_key)
if cached_response:
return cached_response
# 执行请求
import requests
response = requests.request(
method, url,
headers=headers,
params=params
)
# 缓存GET请求结果
if method.upper() == 'GET' and response.status_code == 200:
self.cache.set(cache_key, response.json(), ttl)
return response.json()
@staticmethod
def _make_cache_key(url: str, method: str, params: dict = None) -> str:
"""生成缓存键"""
import json
key_parts = [url, method]
if params:
key_parts.append(json.dumps(params, sort_keys=True))
return "|".join(key_parts)
使用示例
# 创建缓存实例
cache = CacheManager(default_ttl=300, max_size=1000)
# 使用装饰器
@cached(ttl=60, cache=cache)
def expensive_function(user_id: int):
"""模拟耗时操作"""
import time
time.sleep(2) # 模拟耗时
return {"user_id": user_id, "data": "expensive_result"}
# 直接使用
cache.set("my_key", {"value": "test_data"}, ttl=120)
result = cache.get("my_key")
# 使用缓存管理器
cache_manager = CacheManager(strategy=CacheStrategy.LRU)
cache_manager.set("key1", "value1", ttl=60)
cache_manager.delete("key1")
# 获取统计信息
stats = cache_manager.get_stats()
print(f"Cache stats: {stats}")
这个缓存工具封装提供了：
- 多种缓存策略：TTL、LRU、LFU等
- 线程安全：使用锁机制保证并发安全
- 装饰器支持：方便快速缓存函数结果
- 统计功能：监控缓存使用情况
- 可扩展性：支持Redis等外部存储
- 错误处理：完善的日志记录和异常处理
可以根据实际需求选择合适的缓存实现和策略。
