Python脚本如何统计缓存命中率数据

本文目录导读：

1. 基础缓存命中率统计
2. 带时间窗口的统计
3. 完整缓存系统实现
4. 带持久化的统计存储
5. 统计监控和告警
6. 使用建议

基础缓存命中率统计

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分钟检查一次

使用建议

1. 选择合适的统计粒度：根据业务需求选择时间窗口
2. 监控告警设置：设置合理的阈值，及时发现问题
3. 数据持久化：保存历史数据便于分析趋势
4. 性能考虑：统计操作本身不要影响正常缓存性能
5. 多维度统计：可以按缓存类型、Key前缀等维度分别统计

这些实现可以根据你的具体需求进行选择和调整。