本文目录导读：
我来介绍几种监控缓存读写异常情况的Python脚本方法：
基础缓存监控装饰器
import functools
import logging
from datetime import datetime
from typing import Any, Callable
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CacheMonitor:
def __init__(self):
self.stats = {
'hits': 0,
'misses': 0,
'errors': 0,
'total_operations': 0
}
def monitor_cache(self, func: Callable) -> Callable:
"""缓存操作监控装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
self.stats['total_operations'] += 1
cache_key = str(args) + str(kwargs)
try:
result = func(*args, **kwargs)
# 判断是命中还是未命中
if result is not None:
self.stats['hits'] += 1
else:
self.stats['misses'] += 1
return result
except Exception as e:
self.stats['errors'] += 1
logger.error(f"缓存异常: {e} | 时间: {datetime.now()}")
logger.error(f"缓存键: {cache_key}")
raise
return wrapper
# 使用示例
cache_monitor = CacheMonitor()
@cache_monitor.monitor_cache
def get_from_cache(key):
# 模拟缓存读取
pass
@cache_monitor.monitor_cache
def set_to_cache(key, value):
# 模拟缓存写入
pass
Redis缓存监控
import redis
import json
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class CacheMetrics:
operation: str
key: str
status: str
response_time: float
error_message: Optional[str] = None
class RedisCacheMonitor:
def __init__(self, host='localhost', port=6379, db=0):
self.client = redis.Redis(host=host, port=port, db=db)
self.metrics_history = []
self.alert_threshold = 2.0 # 响应时间阈值(秒)
def monitor_get(self, key: str) -> Optional[str]:
"""监控读取操作"""
start_time = time.time()
metric = CacheMetrics(operation='GET', key=key, status='success')
try:
value = self.client.get(key)
response_time = time.time() - start_time
if value is None:
metric.status = 'miss'
else:
metric.status = 'hit'
# 检查响应时间
if response_time > self.alert_threshold:
logger.warning(f"缓存读取慢查询: {key} | 耗时: {response_time:.3f}s")
metric.response_time = response_time
self._record_metric(metric)
return value
except redis.RedisError as e:
metric.status = 'error'
metric.error_message = str(e)
metric.response_time = time.time() - start_time
self._record_metric(metric)
logger.error(f"Redis读取异常: {e}")
raise
def monitor_set(self, key: str, value: str, expire: int = 3600):
"""监控写入操作"""
start_time = time.time()
metric = CacheMetrics(operation='SET', key=key, status='success')
try:
self.client.setex(key, expire, value)
metric.response_time = time.time() - start_time
self._record_metric(metric)
except redis.RedisError as e:
metric.status = 'error'
metric.error_message = str(e)
metric.response_time = time.time() - start_time
self._record_metric(metric)
logger.error(f"Redis写入异常: {e}")
raise
def _record_metric(self, metric: CacheMetrics):
"""记录指标"""
self.metrics_history.append(metric)
# 保持最近的1000条记录
if len(self.metrics_history) > 1000:
self.metrics_history.pop(0)
def get_stats(self):
"""获取统计信息"""
total_ops = len(self.metrics_history)
if not total_ops:
return {}
hits = sum(1 for m in self.metrics_history if m.status == 'hit')
misses = sum(1 for m in self.metrics_history if m.status == 'miss')
errors = sum(1 for m in self.metrics_history if m.status == 'error')
return {
'total_operations': total_ops,
'hit_rate': hits / total_ops if total_ops else 0,
'miss_rate': misses / total_ops if total_ops else 0,
'error_rate': errors / total_ops if total_ops else 0,
'avg_response_time': sum(m.response_time for m in self.metrics_history) / total_ops
}
内存缓存监控
from collections import defaultdict
import threading
import time
from typing import Any
class MonitoredCache:
def __init__(self, max_size=100):
self.cache = {}
self.max_size = max_size
self.max_memory = 100 * 1024 * 1024 # 100MB
self.lock = threading.Lock()
self.stats = defaultdict(int)
self.error_log = []
def get(self, key: str) -> Any:
"""带监控的读取"""
with self.lock:
try:
value = self.cache.get(key)
if value is not None:
self.stats['hits'] += 1
else:
self.stats['misses'] += 1
logger.warning(f"缓存未命中: {key}")
return value
except Exception as e:
self.stats['errors'] += 1
self._log_error('GET', key, str(e))
logger.error(f"缓存读取异常: {key} | 错误: {e}")
return None
def set(self, key: str, value: Any, ttl: int = 300):
"""带监控的写入"""
with self.lock:
try:
# 检查缓存大小
if len(self.cache) >= self.max_size:
logger.warning(f"缓存达到最大容量: {self.max_size}")
# 检查内存使用（简单估计）
estimated_size = len(str(value)) * 2
if estimated_size > 1024 * 1024: # 1MB
logger.warning(f"大对象缓存: {key} | 大小: {estimated_size} bytes")
self.cache[key] = {
'value': value,
'timestamp': time.time(),
'ttl': ttl
}
self.stats['sets'] += 1
except MemoryError as e:
self.stats['memory_errors'] += 1
self._log_error('SET', key, 'MemoryError')
logger.error(f"内存不足，无法缓存: {key}")
# 触发缓存清理
self._cleanup()
except Exception as e:
self.stats['errors'] += 1
self._log_error('SET', key, str(e))
logger.error(f"缓存写入异常: {key} | 错误: {e}")
def _cleanup(self):
"""清理过期缓存"""
current_time = time.time()
expired_keys = []
for key, data in self.cache.items():
if current_time - data['timestamp'] > data['ttl']:
expired_keys.append(key)
for key in expired_keys:
del self.cache[key]
logger.info(f"缓存清理: 移除了 {len(expired_keys)} 个过期条目")
def _log_error(self, operation: str, key: str, error: str):
"""记录错误日志"""
self.error_log.append({
'operation': operation,
'key': key,
'error': error,
'timestamp': time.time()
})
# 保留最近的100条错误记录
if len(self.error_log) > 100:
self.error_log.pop(0)
def check_health(self):
"""健康检查"""
return {
'size': len(self.cache),
'max_size': self.max_size,
'hit_rate': self.stats['hits'] / (self.stats['hits'] + self.stats['misses']) if (self.stats['hits'] + self.stats['misses']) > 0 else 0,
'error_count': self.stats['errors'],
'recent_errors': self.error_log[-5:] if self.error_log else []
}
综合监控系统
import schedule
import time
from typing import Dict, List
class CacheMonitoringSystem:
"""综合缓存监控系统"""
def __init__(self):
self.monitors = {}
self.alerts = []
def add_monitor(self, name: str, monitor):
"""添加缓存监控器"""
self.monitors[name] = monitor
def check_all(self):
"""检查所有缓存系统"""
results = {}
for name, monitor in self.monitors.items():
try:
if hasattr(monitor, 'check_health'):
health = monitor.check_health()
results[name] = health
# 检查异常阈值
if health.get('error_rate', 0) > 0.1: # 错误率超过10%
self._trigger_alert(f"缓存系统 {name} 错误率过高: {health['error_rate']}")
if health.get('hit_rate', 0) < 0.5: # 命中率低于50%
self._trigger_alert(f"缓存系统 {name} 命中率过低: {health['hit_rate']}")
except Exception as e:
logger.error(f"检查缓存系统 {name} 失败: {e}")
results[name] = {'status': 'error', 'error': str(e)}
return results
def _trigger_alert(self, message: str):
"""触发告警"""
alert = {
'message': message,
'timestamp': time.time(),
'level': 'WARNING'
}
self.alerts.append(alert)
logger.warning(f"缓存告警: {message}")
# 这里可以集成邮件、短信等通知
def start_monitoring(self, interval: int = 60):
"""启动定时监控"""
schedule.every(interval).seconds.do(self.check_all)
while True:
schedule.run_pending()
time.sleep(1)
# 使用示例
if __name__ == "__main__":
# 创建监控系统
system = CacheMonitoringSystem()
# 添加各种缓存监控器
system.add_monitor('redis_cache', RedisCacheMonitor())
system.add_monitor('memory_cache', MonitoredCache())
# 启动监控
system.start_monitoring(interval=30) # 每30秒检查一次
异常检测功能
class AnomalyDetector:
"""缓存异常检测器"""
def __init__(self):
self.baseline = {
'avg_response_time': 0.1,
'error_rate': 0.01,
'hit_rate': 0.8
}
self.threshold_multiplier = 3 # 阈值倍数
def detect_anomaly(self, metrics: Dict) -> List[str]:
"""检测异常"""
anomalies = []
# 检查响应时间异常
if metrics['avg_response_time'] > self.baseline['avg_response_time'] * self.threshold_multiplier:
anomalies.append(f"响应时间异常: {metrics['avg_response_time']:.3f}s (基线: {self.baseline['avg_response_time']:.3f}s)")
# 检查错误率异常
if metrics['error_rate'] > self.baseline['error_rate'] * self.threshold_multiplier:
anomalies.append(f"错误率异常: {metrics['error_rate']:.2%} (基线: {self.baseline['error_rate']:.2%})")
# 检查命中率异常
if metrics['hit_rate'] < self.baseline['hit_rate'] / self.threshold_multiplier:
anomalies.append(f"命中率异常: {metrics['hit_rate']:.2%} (基线: {self.baseline['hit_rate']:.2%})")
return anomalies
通用异常处理
def safe_cache_operation(func):
"""缓存操作安全执行器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
max_retries = 3
retry_delay = 0.1
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except redis.ConnectionError as e:
if attempt == max_retries - 1:
logger.error(f"缓存连接失败，重试耗尽: {e}")
return None
logger.warning(f"缓存连接重试 #{attempt + 1}")
time.sleep(retry_delay * (attempt + 1))
except redis.TimeoutError:
logger.warning("缓存超时")
return None
except Exception as e:
logger.error(f"缓存操作异常: {e}")
return None
return wrapper
# 使用示例
@safe_cache_operation
def get_user_data(user_id: str):
# 获取缓存数据的逻辑
pass
这些脚本覆盖了：
- 基础监控：记录命中率、错误率等统计信息
- Redis监控：针对Redis的专门监控
- 内存缓存监控：内存使用、过期清理等
- 告警系统：异常检测和告警
- 性能监控：响应时间、超时等
- 自动恢复：重试机制、降级策略
根据实际需求,可以选择合适的监控级别和告警方式。
