本文目录导读：
我来详细说明如何用Python脚本筛选和保留有效缓存数据。
基本缓存筛选策略
import time
import json
from datetime import datetime, timedelta
from typing import Dict, Any, Optional
class CacheManager:
"""缓存管理器"""
def __init__(self, max_size: int = 1000, ttl: int = 3600):
self.cache: Dict[str, Dict[str, Any]] = {}
self.max_size = max_size
self.ttl = ttl # 默认过期时间（秒）
def set(self, key: str, value: Any, ttl: Optional[int] = None):
"""设置缓存"""
if len(self.cache) >= self.max_size:
self._evict_expired()
self.cache[key] = {
'value': value,
'timestamp': time.time(),
'ttl': ttl or self.ttl,
'access_count': 0,
'last_access': time.time()
}
def get(self, key: str) -> Optional[Any]:
"""获取缓存"""
if key not in self.cache:
return None
entry = self.cache[key]
# 检查是否过期
if self._is_expired(entry):
del self.cache[key]
return None
# 更新访问信息
entry['access_count'] += 1
entry['last_access'] = time.time()
return entry['value']
def _is_expired(self, entry: Dict) -> bool:
"""检查缓存是否过期"""
return time.time() - entry['timestamp'] > entry['ttl']
def _evict_expired(self):
"""清除过期缓存"""
expired_keys = [
key for key, entry in self.cache.items()
if self._is_expired(entry)
]
for key in expired_keys:
del self.cache[key]
def get_valid_cache(self) -> Dict[str, Any]:
"""获取所有有效缓存"""
self._evict_expired()
return {
key: entry['value']
for key, entry in self.cache.items()
}
def filter_cache(self,
min_access_count: int = 0,
max_age: Optional[int] = None,
custom_filter: Optional[callable] = None) -> Dict[str, Any]:
"""筛选有效缓存数据"""
self._evict_expired()
filtered = {}
for key, entry in self.cache.items():
# 基于访问次数筛选
if entry['access_count'] < min_access_count:
continue
# 基于年龄筛选
if max_age and (time.time() - entry['timestamp'] > max_age):
continue
# 自定义筛选条件
if custom_filter and not custom_filter(key, entry):
continue
filtered[key] = entry['value']
return filtered
基于策略的缓存筛选
from enum import Enum
from collections import defaultdict
class EvictionPolicy(Enum):
"""淘汰策略"""
LRU = "least_recently_used"
LFU = "least_frequently_used"
FIFO = "first_in_first_out"
TTL = "time_to_live"
class StrategyCacheManager:
"""策略驱动缓存管理器"""
def __init__(self, max_size: int = 1000):
self.cache = {}
self.max_size = max_size
self.access_history = []
self.frequency = defaultdict(int)
def set(self, key: str, value: Any, ttl: int = 3600):
"""设置缓存"""
if len(self.cache) >= self.max_size:
self._evict_by_strategy()
self.cache[key] = {
'value': value,
'created_at': time.time(),
'ttl': ttl,
'last_access': time.time()
}
def get(self, key: str) -> Optional[Any]:
"""获取缓存"""
if key not in self.cache:
return None
entry = self.cache[key]
# 检查TTL
if time.time() - entry['created_at'] > entry['ttl']:
del self.cache[key]
return None
# 更新访问记录
entry['last_access'] = time.time()
self.frequency[key] += 1
self.access_history.append((time.time(), key))
return entry['value']
def _evict_by_strategy(self, policy: EvictionPolicy = EvictionPolicy.LRU):
"""根据策略淘汰缓存"""
if not self.cache:
return
if policy == EvictionPolicy.LRU:
# 淘汰最久未访问的
victim = min(self.cache.items(),
key=lambda x: x[1]['last_access'])
del self.cache[victim[0]]
elif policy == EvictionPolicy.LFU:
# 淘汰访问频率最低的
victim = min(self.cache.keys(),
key=lambda k: self.frequency.get(k, 0))
del self.cache[victim]
elif policy == EvictionPolicy.FIFO:
# 淘汰最早创建的
victim = min(self.cache.items(),
key=lambda x: x[1]['created_at'])
del self.cache[victim[0]]
elif policy == EvictionPolicy.TTL:
# 淘汰即将过期的
victim = min(self.cache.items(),
key=lambda x: x[1]['ttl'] -
(time.time() - x[1]['created_at']))
del self.cache[victim[0]]
def filter_by_policy(self,
policy: EvictionPolicy,
keep_percentage: float = 0.8) -> Dict[str, Any]:
"""基于策略筛选保留的数据"""
if not self.cache:
return {}
# 计算要保留的数量
keep_count = int(len(self.cache) * keep_percentage)
# 根据策略排序
if policy == EvictionPolicy.LRU:
sorted_keys = sorted(
self.cache.keys(),
key=lambda k: self.cache[k]['last_access'],
reverse=True
)
elif policy == EvictionPolicy.LFU:
sorted_keys = sorted(
self.cache.keys(),
key=lambda k: self.frequency.get(k, 0),
reverse=True
)
elif policy == EvictionPolicy.FIFO:
sorted_keys = sorted(
self.cache.keys(),
key=lambda k: self.cache[k]['created_at'],
reverse=True
)
else:
sorted_keys = list(self.cache.keys())
# 保留排名靠前的
kept_keys = sorted_keys[:keep_count]
return {k: self.cache[k]['value'] for k in kept_keys}
基于规则的缓存筛选
class RuleBasedCacheFilter:
"""基于规则的缓存筛选器"""
def __init__(self):
self.rules = []
def add_rule(self,
name: str,
condition: callable,
action: str = 'keep',
priority: int = 0):
"""添加筛选规则"""
self.rules.append({
'name': name,
'condition': condition,
'action': action, # 'keep' or 'remove'
'priority': priority
})
self.rules.sort(key=lambda x: x['priority'], reverse=True)
def apply_rules(self, cache: Dict[str, Any]) -> Dict[str, Any]:
"""应用规则筛选缓存"""
result = {}
for key, value in cache.items():
action = None
for rule in self.rules:
if rule['condition'](key, value):
action = rule['action']
break
if action == 'keep' or action is None:
result[key] = value
return result
# 使用示例
filter = RuleBasedCacheFilter()
# 添加规则
filter.add_rule(
name="high_frequency",
condition=lambda k, v: v.get('frequency', 0) > 100,
action='keep',
priority=10
)
filter.add_rule(
name="expired",
condition=lambda k, v: v.get('expired', False),
action='remove',
priority=5
)
# 应用规则
filtered_cache = filter.apply_rules(cache_data)
高级缓存筛选器
import hashlib
import pickle
from typing import List, Tuple
class AdvancedCacheFilter:
"""高级缓存筛选器"""
def __init__(self):
self.filters = []
def add_filter(self, filter_func: callable, name: str = None):
"""添加筛选函数"""
self.filters.append((name or f"filter_{len(self.filters)}", filter_func))
def filter_by_pattern(self,
cache: Dict[str, Any],
patterns: List[str]) -> Dict[str, Any]:
"""基于模式筛选"""
import re
result = {}
for key, value in cache.items():
for pattern in patterns:
if re.match(pattern, key):
result[key] = value
break
return result
def filter_by_size(self,
cache: Dict[str, Any],
max_size: int,
size_func: callable = None) -> Dict[str, Any]:
"""基于大小筛选"""
if size_func is None:
size_func = lambda v: len(pickle.dumps(v))
sorted_items = sorted(
cache.items(),
key=lambda x: size_func(x[1]),
reverse=True
)
result = {}
current_size = 0
for key, value in sorted_items:
item_size = size_func(value)
if current_size + item_size <= max_size:
result[key] = value
current_size += item_size
else:
break
return result
def filter_by_hash(self,
cache: Dict[str, Any],
algorithm: str = 'md5') -> Dict[str, Any]:
"""基于哈希去重筛选"""
seen_hashes = set()
result = {}
for key, value in cache.items():
content = pickle.dumps(value)
hash_obj = hashlib.new(algorithm, content)
content_hash = hash_obj.hexdigest()
if content_hash not in seen_hashes:
seen_hashes.add(content_hash)
result[key] = value
return result
def apply_all_filters(self, cache: Dict[str, Any]) -> Dict[str, Any]:
"""应用所有筛选器"""
result = cache.copy()
for name, filter_func in self.filters:
result = filter_func(result)
print(f"Applied filter '{name}': {len(result)} items remaining")
return result
# 使用示例
filter = AdvancedCacheFilter()
# 添加多个筛选器
filter.add_filter(lambda c: filter.filter_by_pattern(c, [r'^user_.*']), "user_filter")
filter.add_filter(lambda c: filter.filter_by_size(c, 1024*1024), "size_filter")
filter.add_filter(lambda c: filter.filter_by_hash(c), "hash_filter")
# 应用所有筛选
final_cache = filter.apply_all_filters(cache_data)
监控和统计筛选
class CacheAnalyzer:
"""缓存分析器"""
@staticmethod
def analyze_cache(cache: Dict[str, Any]) -> Dict:
"""分析缓存健康状况"""
if not cache:
return {'status': 'empty', 'message': 'Cache is empty'}
total_size = 0
sizes = []
ages = []
access_counts = []
for key, entry in cache.items():
if isinstance(entry, dict) and 'value' in entry:
value = entry['value']
else:
value = entry
# 计算大小
size = len(pickle.dumps(value))
total_size += size
sizes.append(size)
# 计算年龄
if isinstance(entry, dict) and 'timestamp' in entry:
age = time.time() - entry['timestamp']
ages.append(age)
# 访问次数
if isinstance(entry, dict) and 'access_count' in entry:
access_counts.append(entry['access_count'])
# 生成统计信息
stats = {
'total_items': len(cache),
'total_size_bytes': total_size,
'total_size_mb': total_size / 1024 / 1024,
'avg_size': sum(sizes) / len(sizes) if sizes else 0,
'max_size': max(sizes) if sizes else 0,
'min_size': min(sizes) if sizes else 0,
'avg_age': sum(ages) / len(ages) if ages else 0,
'max_age': max(ages) if ages else 0,
'total_accesses': sum(access_counts) if access_counts else 0,
'avg_accesses': sum(access_counts) / len(access_counts) if access_counts else 0,
'status': 'healthy'
}
return stats
@staticmethod
def suggest_retention(cache: Dict[str, Any]) -> List[str]:
"""建议保留的缓存项"""
stats = CacheAnalyzer.analyze_cache(cache)
suggestions = []
# 基于访问频率
if stats['avg_accesses'] > 0:
threshold = stats['avg_accesses'] * 0.1
low_access = [
k for k, v in cache.items()
if isinstance(v, dict) and
v.get('access_count', 0) < threshold
]
suggestions.extend(low_access[:10]) # 最多建议10个
# 基于年龄
if stats['max_age'] > 86400: # 超过1天
old_items = [
k for k, v in cache.items()
if isinstance(v, dict) and
time.time() - v.get('timestamp', 0) > 86400
]
suggestions.extend(old_items[:10])
return suggestions
完整使用示例
def main():
"""完整示例"""
# 创建缓存管理器
cache = CacheManager(max_size=1000, ttl=3600)
# 添加一些测试数据
for i in range(100):
cache.set(f"key_{i}", {
'data': f"value_{i}",
'timestamp': time.time() - random.randint(0, 7200)
})
# 模拟访问
for _ in range(50):
key = f"key_{random.randint(0, 99)}"
cache.get(key)
# 筛选有效缓存
print("=== 基础筛选 ===")
valid_cache = cache.get_valid_cache()
print(f"有效缓存数: {len(valid_cache)}")
# 高级筛选
print("\n=== 高级筛选 ===")
filtered = cache.filter_cache(
min_access_count=2, # 至少访问2次
max_age=1800, # 不超过30分钟
custom_filter=lambda k, v: v['ttl'] > 1800 # TTL大于30分钟
)
print(f"筛选后缓存数: {len(filtered)}")
# 分析缓存
print("\n=== 缓存分析 ===")
analyzer = CacheAnalyzer()
stats = analyzer.analyze_cache(cache.cache)
print(f"缓存统计: {json.dumps(stats, indent=2)}")
# 获取保留建议
suggestions = analyzer.suggest_retention(cache.cache)
print(f"\n建议保留的键: {suggestions[:5]}")
if __name__ == "__main__":
import random
main()
这些方法可以根据你的具体需求选择合适的缓存筛选策略，关键是要根据实际使用场景（如访问频率、数据大小、时效性等）来制定筛选规则。
