本文目录导读：
我来介绍几种监控缓存读写异常情况的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的专门监控
- 内存缓存监控：内存使用、过期清理等
- 告警系统：异常检测和告警
- 性能监控：响应时间、超时等
- 自动恢复：重试机制、降级策略
根据实际需求,可以选择合适的监控级别和告警方式。
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
