本文目录导读：
基础缓存命中率统计
class CacheStats:
def __init__(self):
self.hits = 0
self.misses = 0
def record_hit(self):
self.hits += 1
def record_miss(self):
self.misses += 1
def get_hit_rate(self):
total = self.hits + self.misses
if total == 0:
return 0.0
return self.hits / total * 100
def get_stats(self):
return {
'hits': self.hits,
'misses': self.misses,
'total': self.hits + self.misses,
'hit_rate': f"{self.get_hit_rate():.2f}%"
}
# 使用示例
cache_stats = CacheStats()
# 模拟缓存操作
def get_from_cache(key):
# 模拟缓存命中/未命中
if key in simulated_cache:
cache_stats.record_hit()
return simulated_cache[key]
else:
cache_stats.record_miss()
return None
带时间窗口的统计
from collections import deque
import time
from typing import Any, Optional
class TimeWindowCacheStats:
def __init__(self, window_seconds: int = 3600):
self.window_seconds = window_seconds
self.hits = deque()
self.misses = deque()
def record_hit(self):
self.hits.append(time.time())
self._cleanup()
def record_miss(self):
self.misses.append(time.time())
self._cleanup()
def _cleanup(self):
"""清理过期记录"""
current_time = time.time()
cutoff = current_time - self.window_seconds
while self.hits and self.hits[0] < cutoff:
self.hits.popleft()
while self.misses and self.misses[0] < cutoff:
self.misses.popleft()
def get_hit_rate(self) -> float:
self._cleanup()
total = len(self.hits) + len(self.misses)
if total == 0:
return 0.0
return len(self.hits) / total * 100
def get_stats(self) -> dict:
return {
'hits': len(self.hits),
'misses': len(self.misses),
'total': len(self.hits) + len(self.misses),
'hit_rate': f"{self.get_hit_rate():.2f}%",
'window_seconds': self.window_seconds
}
完整缓存系统实现
import redis
from functools import wraps
import json
from datetime import datetime
import threading
class AdvancedCacheStats:
def __init__(self, redis_client=None):
self.stats_lock = threading.Lock()
self.cache_stats = {
'hits': 0,
'misses': 0,
'expired': 0,
'error': 0,
'start_time': datetime.now()
}
self.redis_client = redis_client or self._create_local_cache()
def _create_local_cache(self):
"""创建本地缓存（无Redis时的后备方案）"""
class LocalCache:
def __init__(self):
self._cache = {}
def get(self, key):
return self._cache.get(key)
def set(self, key, value, ex=None):
self._cache[key] = value
return LocalCache()
def record_hit(self):
with self.stats_lock:
self.cache_stats['hits'] += 1
def record_miss(self, reason='miss'):
with self.stats_lock:
if reason in self.cache_stats:
self.cache_stats[reason] += 1
else:
self.cache_stats['misses'] += 1
def get_stats(self):
with self.stats_lock:
total = self.cache_stats['hits'] + self.cache_stats['misses']
hit_rate = (self.cache_stats['hits'] / total * 100) if total > 0 else 0
return {
**self.cache_stats,
'total_requests': total,
'hit_rate': f"{hit_rate:.2f}%",
'uptime': str(datetime.now() - self.cache_stats['start_time'])
}
def cached(self, ttl=300):
"""缓存装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 生成缓存键
cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
# 尝试获取缓存
cached_value = self.redis_client.get(cache_key)
if cached_value is not None:
self.record_hit()
return json.loads(cached_value)
# 缓存未命中，执行函数
self.record_miss()
result = func(*args, **kwargs)
# 设置缓存
try:
self.redis_client.set(
cache_key,
json.dumps(result),
ex=ttl
)
except Exception as e:
self.record_miss('error')
print(f"Cache set error: {e}")
return result
return wrapper
return decorator
# 使用示例
cache_stats = AdvancedCacheStats()
@cache_stats.cached(ttl=60)
def expensive_operation(param1, param2):
"""模拟耗时操作"""
import time
time.sleep(1)
return {
'result': param1 + param2,
'timestamp': datetime.now().isoformat()
}
# 测试
for i in range(10):
expensive_operation(1, 2)
expensive_operation(3, 4)
print(cache_stats.get_stats())
带持久化的统计存储
import sqlite3
from datetime import datetime
import json
class PersistentCacheStats:
def __init__(self, db_path='cache_stats.db'):
self.conn = sqlite3.connect(db_path)
self._create_tables()
def _create_tables(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS cache_stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
hits INTEGER DEFAULT 0,
misses INTEGER DEFAULT 0,
total_requests INTEGER DEFAULT 0,
hit_rate REAL DEFAULT 0.0
)
''')
self.conn.commit()
def log_stats(self, hits, misses):
total = hits + misses
hit_rate = (hits / total * 100) if total > 0 else 0
cursor = self.conn.cursor()
cursor.execute('''
INSERT INTO cache_stats
(hits, misses, total_requests, hit_rate)
VALUES (?, ?, ?, ?)
''', (hits, misses, total, hit_rate))
self.conn.commit()
def get_history(self, limit=100):
cursor = self.conn.cursor()
cursor.execute('''
SELECT * FROM cache_stats
ORDER BY timestamp DESC
LIMIT ?
''', (limit,))
columns = [description[0] for description in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
def get_average_hit_rate(self, hours=24):
cursor = self.conn.cursor()
cursor.execute('''
SELECT AVG(hit_rate) as avg_hit_rate,
SUM(hits) as total_hits,
SUM(misses) as total_misses
FROM cache_stats
WHERE timestamp >= datetime('now', '-? hours')
''', (hours,))
result = cursor.fetchone()
return {
'average_hit_rate': f"{result[0]:.2f}%" if result[0] else "0.00%",
'total_hits': result[1] or 0,
'total_misses': result[2] or 0
}
统计监控和告警
class CacheMonitor:
def __init__(self, stats_collector, threshold=80.0):
self.stats = stats_collector
self.threshold = threshold
self.alert_callbacks = []
def add_alert_callback(self, callback):
self.alert_callbacks.append(callback)
def check_and_alert(self):
stats = self.stats.get_stats()
hit_rate = float(stats['hit_rate'].rstrip('%'))
if hit_rate < self.threshold:
alert_msg = f"Cache hit rate {hit_rate:.2f}% below threshold {self.threshold}%"
print(f"⚠️ ALERT: {alert_msg}")
for callback in self.alert_callbacks:
callback(alert_msg, stats)
def start_monitoring(self, interval=60):
"""启动定期监控"""
import threading
def monitor_loop():
while True:
self.check_and_alert()
threading.Event().wait(interval)
thread = threading.Thread(target=monitor_loop, daemon=True)
thread.start()
# 使用示例
def send_alert_to_slack(message, stats):
"""发送告警到Slack"""
print(f"Slack alert: {message}")
print(f"Stats: {json.dumps(stats, indent=2)}")
# 创建监控
monitor = CacheMonitor(cache_stats, threshold=90.0)
monitor.add_alert_callback(send_alert_to_slack)
monitor.start_monitoring(interval=300) # 每5分钟检查一次
使用建议
- 选择合适的统计粒度：根据业务需求选择时间窗口
- 监控告警设置：设置合理的阈值，及时发现问题
- 数据持久化：保存历史数据便于分析趋势
- 性能考虑：统计操作本身不要影响正常缓存性能
- 多维度统计：可以按缓存类型、Key前缀等维度分别统计
这些实现可以根据你的具体需求进行选择和调整。
