Python脚本如何排查缓存读写异常问题

本文目录导读：

1. 基础缓存读写监控脚本
2. 具体缓存实现监控
3. 异常分析工具类
4. 使用示例
5. 关键排查点

我来帮你分析Python排查缓存读写异常的方法和脚本示例。

基础缓存读写监控脚本

#!/usr/bin/env python3
"""
缓存读写异常排查工具
支持: Redis、Memcached、本地缓存
"""
import time
import logging
import traceback
from typing import Any, Optional
from datetime import datetime
from functools import wraps
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('cache_monitor.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)
class CacheMonitor:
    """缓存监控器"""
    def __init__(self, name: str = "default"):
        self.name = name
        self.stats = {
            'read_attempts': 0,
            'write_attempts': 0,
            'read_success': 0,
            'write_success': 0,
            'read_failures': 0,
            'write_failures': 0,
            'total_read_time': 0,
            'total_write_time': 0,
            'last_error': None,
            'error_count': 0
        }
        self.slow_threshold = 1.0  # 慢操作阈值（秒）
    def monitor_read(self, func):
        """装饰器：监控读操作"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            self.stats['read_attempts'] += 1
            start_time = time.time()
            try:
                result = func(*args, **kwargs)
                elapsed = time.time() - start_time
                self.stats['read_success'] += 1
                self.stats['total_read_time'] += elapsed
                # 记录慢操作
                if elapsed > self.slow_threshold:
                    logger.warning(
                        f"慢读操作: {func.__name__} 耗时 {elapsed:.3f}s"
                    )
                return result
            except Exception as e:
                elapsed = time.time() - start_time
                self.stats['read_failures'] += 1
                self.stats['last_error'] = str(e)
                self.stats['error_count'] += 1
                logger.error(
                    f"缓存读取异常: {func.__name__}\n"
                    f"错误信息: {str(e)}\n"
                    f"堆栈: {traceback.format_exc()}"
                )
                raise
        return wrapper
    def monitor_write(self, func):
        """装饰器：监控写操作"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            self.stats['write_attempts'] += 1
            start_time = time.time()
            try:
                result = func(*args, **kwargs)
                elapsed = time.time() - start_time
                self.stats['write_success'] += 1
                self.stats['total_write_time'] += elapsed
                if elapsed > self.slow_threshold:
                    logger.warning(
                        f"慢写操作: {func.__name__} 耗时 {elapsed:.3f}s"
                    )
                return result
            except Exception as e:
                elapsed = time.time() - start_time
                self.stats['write_failures'] += 1
                self.stats['last_error'] = str(e)
                self.stats['error_count'] += 1
                logger.error(
                    f"缓存写入异常: {func.__name__}\n"
                    f"错误信息: {str(e)}\n"
                    f"堆栈: {traceback.format_exc()}"
                )
                raise
        return wrapper
    def get_report(self) -> dict:
        """获取监控报告"""
        total_ops = self.stats['read_attempts'] + self.stats['write_attempts']
        read_fail_rate = (
            self.stats['read_failures'] / self.stats['read_attempts'] * 100
            if self.stats['read_attempts'] > 0 else 0
        )
        write_fail_rate = (
            self.stats['write_failures'] / self.stats['write_attempts'] * 100
            if self.stats['write_attempts'] > 0 else 0
        )
        avg_read_time = (
            self.stats['total_read_time'] / self.stats['read_success']
            if self.stats['read_success'] > 0 else 0
        )
        avg_write_time = (
            self.stats['total_write_time'] / self.stats['write_success']
            if self.stats['write_success'] > 0 else 0
        )
        return {
            'monitor_name': self.name,
            'total_operations': total_ops,
            'read_stats': {
                'attempts': self.stats['read_attempts'],
                'success': self.stats['read_success'],
                'failures': self.stats['read_failures'],
                'failure_rate': f"{read_fail_rate:.2f}%",
                'avg_time': f"{avg_read_time:.3f}s"
            },
            'write_stats': {
                'attempts': self.stats['write_attempts'],
                'success': self.stats['write_success'],
                'failures': self.stats['write_failures'],
                'failure_rate': f"{write_fail_rate:.2f}%",
                'avg_time': f"{avg_write_time:.3f}s"
            },
            'errors': {
                'total_errors': self.stats['error_count'],
                'last_error': self.stats['last_error']
            },
            'timestamp': datetime.now().isoformat()
        }

具体缓存实现监控

import redis
import json
from typing import Optional
class RedisCacheMonitor:
    """Redis缓存监控"""
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis_client = redis.Redis(
            host=host, 
            port=port, 
            db=db,
            socket_connect_timeout=5,
            socket_timeout=5,
            retry_on_timeout=True,
            health_check_interval=30
        )
        self.monitor = CacheMonitor("redis")
        self.connection_errors = 0
        self.timeout_errors = 0
    def check_connection(self) -> bool:
        """检查Redis连接"""
        try:
            self.redis_client.ping()
            return True
        except redis.ConnectionError as e:
            logger.error(f"Redis连接失败: {e}")
            self.connection_errors += 1
            return False
        except redis.TimeoutError as e:
            logger.error(f"Redis连接超时: {e}")
            self.timeout_errors += 1
            return False
    @CacheMonitor.monitor_read
    def get(self, key: str) -> Optional[Any]:
        """读取缓存"""
        if not self.check_connection():
            return None
        try:
            data = self.redis_client.get(key)
            if data:
                return json.loads(data)
            return None
        except redis.RedisError as e:
            logger.error(f"Redis读取异常 [{key}]: {e}")
            raise
    @CacheMonitor.monitor_write
    def set(self, key: str, value: Any, expire: int = 3600):
        """写入缓存"""
        if not self.check_connection():
            return False
        try:
            serialized = json.dumps(value, default=str)
            return self.redis_client.setex(key, expire, serialized)
        except redis.RedisError as e:
            logger.error(f"Redis写入异常 [{key}]: {e}")
            raise
    def health_check(self) -> dict:
        """健康检查"""
        return {
            'connection_ok': self.check_connection(),
            'connection_errors': self.connection_errors,
            'timeout_errors': self.timeout_errors,
            'memory_info': self._get_memory_info()
        }
    def _get_memory_info(self) -> dict:
        """获取内存信息"""
        try:
            info = self.redis_client.info('memory')
            return {
                'used_memory': info.get('used_memory_human', 'N/A'),
                'peak_memory': info.get('used_memory_peak_human', 'N/A'),
                'fragmentation': info.get('mem_fragmentation_ratio', 'N/A')
            }
        except:
            return {'error': '无法获取内存信息'}

异常分析工具类

import statistics
from collections import deque
from typing import List, Dict
class CacheExceptionAnalyzer:
    """缓存异常分析器"""
    def __init__(self, window_size: int = 100):
        self.window_size = window_size
        self.exception_history = deque(maxlen=window_size)
        self.time_buckets = {}  # 按时间段统计异常
    def record_exception(self, exception_type: str, key: str, 
                        error_msg: str, duration: float):
        """记录异常"""
        record = {
            'timestamp': time.time(),
            'type': exception_type,
            'key': key,
            'message': error_msg,
            'duration': duration
        }
        self.exception_history.append(record)
        # 按小时统计
        hour_key = datetime.now().strftime('%Y-%m-%d %H:00')
        if hour_key not in self.time_buckets:
            self.time_buckets[hour_key] = []
        self.time_buckets[hour_key].append(record)
    def analyze_exceptions(self) -> Dict:
        """分析异常模式"""
        if not self.exception_history:
            return {'message': '没有异常记录'}
        # 异常类型统计
        type_stats = {}
        for record in self.exception_history:
            exc_type = record['type']
            type_stats[exc_type] = type_stats.get(exc_type, 0) + 1
        # 频率分析
        recent_count = len(self.exception_history)
        time_span = (time.time() - self.exception_history[0]['timestamp']) / 60  # 分钟
        # 识别异常模式
        patterns = []
        # 检查是否频繁超时
        timeout_records = [
            r for r in self.exception_history 
            if 'timeout' in r['type'].lower()
        ]
        if len(timeout_records) > self.window_size * 0.3:
            patterns.append("频繁超时 - 建议检查网络和连接池")
        # 检查连接错误
        conn_errors = [
            r for r in self.exception_history 
            if 'connection' in r['type'].lower()
        ]
        if conn_errors:
            patterns.append("存在连接错误 - 建议检查Redis/Memcached服务器状态")
        # 检查键冲突
        key_conflicts = [
            r for r in self.exception_history 
            if 'key' in r['type'].lower()
        ]
        if key_conflicts:
            patterns.append("键相关错误 - 建议检查键命名规范")
        return {
            'total_exceptions': recent_count,
            'frequency_per_minute': f"{recent_count/time_span:.2f}" if time_span > 0 else "0",
            'type_distribution': type_stats,
            'detected_patterns': patterns,
            'top_error_keys': self._get_top_error_keys(5),
            'time_based_distribution': self.time_buckets
        }
    def _get_top_error_keys(self, n: int) -> List:
        """获取最常见的异常键"""
        key_count = {}
        for record in self.exception_history:
            key = record['key']
            key_count[key] = key_count.get(key, 0) + 1
        return sorted(
            key_count.items(), 
            key=lambda x: x[1], 
            reverse=True
        )[:n]

使用示例

def main():
    """主函数示例"""
    # 初始化缓存监控
    redis_monitor = RedisCacheMonitor(
        host='localhost',
        port=6379
    )
    analyzer = CacheExceptionAnalyzer(window_size=1000)
    # 模拟业务操作
    try:
        # 写入缓存
        redis_monitor.set('user:1001', {
            'name': '张三',
            'age': 30,
            'last_login': datetime.now().isoformat()
        }, expire=3600)
        # 读取缓存
        user_data = redis_monitor.get('user:1001')
        print(f"读取到数据: {user_data}")
    except redis.ConnectionError:
        analyzer.record_exception(
            'ConnectionError',
            'user:1001',
            'Redis连接失败',
            time.time()
        )
    except redis.TimeoutError:
        analyzer.record_exception(
            'TimeoutError',
            'user:1001',
            '读取超时',
            time.time()
        )
    finally:
        # 获取监控报告
        report = redis_monitor.monitor.get_report()
        print("缓存监控报告:", json.dumps(report, indent=2, ensure_ascii=False))
        # 异常分析
        analysis = analyzer.analyze_exceptions()
        if analysis:
            print("异常分析结果:", json.dumps(analysis, indent=2, ensure_ascii=False))
        # 健康检查
        health = redis_monitor.health_check()
        print("健康检查:", json.dumps(health, indent=2, ensure_ascii=False))
if __name__ == "__main__":
    main()

关键排查点

这个脚本可以帮助你：

1. 实时监控所有缓存操作
2. 自动分析异常模式
3. 生成报告便于定位问题
4. 健康检查及时发现潜在风险

建议定期运行此脚本,并结合日志监控系统进行持续观察。