本文目录导读：
我来帮你实现一个完整的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等外部存储
- 错误处理：完善的日志记录和异常处理
可以根据实际需求选择合适的缓存实现和策略。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
