监控缓存命中率的脚本如何写

wen 实用脚本 2026-07-10 2

本文目录导读：

监控缓存命中率的脚本如何写

Redis 缓存命中率监控
Memcached 缓存命中率监控
应用层缓存监控(通用版本)
Nginx 缓存命中率监控
Prometheus + Grafana 监控
使用建议

Redis 缓存命中率监控

Bash 脚本

#!/bin/bash
# Redis缓存命中率监控脚本
REDIS_HOST="localhost"
REDIS_PORT=6379
REDIS_PASSWORD=""
# 获取Redis信息
get_redis_stats() {
    if [ -n "$REDIS_PASSWORD" ]; then
        redis_info=$(redis-cli -h $REDIS_HOST -p $REDIS_PORT -a $REDIS_PASSWORD info stats 2>/dev/null)
    else
        redis_info=$(redis-cli -h $REDIS_HOST -p $REDIS_PORT info stats 2>/dev/null)
    fi
    hits=$(echo "$redis_info" | grep "keyspace_hits:" | awk -F: '{print $2}' | tr -d '\r')
    misses=$(echo "$redis_info" | grep "keyspace_misses:" | awk -F: '{print $2}' | tr -d '\r')
    if [ -n "$hits" ] && [ -n "$misses" ]; then
        total=$((hits + misses))
        if [ $total -gt 0 ]; then
            hit_rate=$(echo "scale=2; $hits * 100 / $total" | bc)
            echo "缓存命中率: ${hit_rate}%"
            echo "总请求数: $total"
            echo "命中次数: $hits"
            echo "未命中次数: $misses"
        else
            echo "暂无缓存请求数据"
        fi
    else
        echo "无法获取Redis统计信息"
    fi
}
# 持续监控
while true; do
    clear
    echo "=== Redis 缓存命中率监控 ==="
    echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
    echo "------------------------"
    get_redis_stats
    echo "------------------------"
    sleep 5
done

Python 脚本（更灵活）

#!/usr/bin/env python3
import redis
import time
from datetime import datetime
class CacheMonitor:
    def __init__(self, host='localhost', port=6379, password=None):
        self.client = redis.Redis(
            host=host,
            port=port,
            password=password,
            decode_responses=True
        )
        self.hits_prev = 0
        self.misses_prev = 0
    def get_hit_rate(self):
        """获取当前命中率"""
        info = self.client.info('stats')
        hits = info.get('keyspace_hits', 0)
        misses = info.get('keyspace_misses', 0)
        # 计算时间段内的命中率
        hits_diff = hits - self.hits_prev
        misses_diff = misses - self.misses_prev
        total_diff = hits_diff + misses_diff
        if total_diff > 0:
            hit_rate = (hits_diff / total_diff) * 100
            return {
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'hit_rate': round(hit_rate, 2),
                'total_hits': hits,
                'total_misses': misses,
                'period_hits': hits_diff,
                'period_misses': misses_diff,
                'period_total': total_diff
            }
        self.hits_prev = hits
        self.misses_prev = misses
        return None
    def monitor(self, interval=5, threshold=80):
        """持续监控，低于阈值报警"""
        print("开始监控Redis缓存命中率...")
        print(f"监控间隔: {interval}秒")
        print(f"告警阈值: {threshold}%")
        print("-" * 60)
        while True:
            stats = self.get_hit_rate()
            if stats:
                print(f"[{stats['timestamp']}] "
                      f"命中率: {stats['hit_rate']}% | "
                      f"总请求: {stats['period_total']} | "
                      f"命中: {stats['period_hits']} | "
                      f"未命中: {stats['period_misses']}")
                if stats['hit_rate'] < threshold:
                    print(f"⚠️ 警告: 命中率低于阈值({threshold}%)!")
            time.sleep(interval)
# 使用示例
if __name__ == "__main__":
    monitor = CacheMonitor()
    monitor.monitor(interval=5, threshold=80)

Memcached 缓存命中率监控

Bash 脚本

#!/bin/bash
# Memcached缓存命中率监控
MEMCACHED_HOST="localhost"
MEMCACHED_PORT=11211
get_memcached_stats() {
    # 使用echo发送stats命令到memcached
    stats=$(echo -e "stats\r" | nc -w 1 $MEMCACHED_HOST $MEMCACHED_PORT 2>/dev/null)
    hits=$(echo "$stats" | grep "STAT get_hits" | awk '{print $3}')
    misses=$(echo "$stats" | grep "STAT get_misses" | awk '{print $3}')
    cmd_get=$(echo "$stats" | grep "STAT cmd_get" | awk '{print $3}')
    if [ -n "$hits" ] && [ -n "$misses" ]; then
        total=$((hits + misses))
        if [ $total -gt 0 ]; then
            hit_rate=$(echo "scale=2; $hits * 100 / $total" | bc)
            echo "Memcached 缓存状态:"
            echo "命中率: ${hit_rate}%"
            echo "Get请求数: $cmd_get"
            echo "命中次数: $hits"
            echo "未命中次数: $misses"
        fi
    fi
}
# 每5秒监控一次
while true; do
    clear
    echo "=== Memcached 缓存命中率监控 ==="
    date
    echo "----------------------------"
    get_memcached_stats
    echo "----------------------------"
    sleep 5
done

应用层缓存监控(通用版本)

Python 装饰器模式

import functools
import time
from collections import defaultdict
class CacheMonitor:
    """应用层缓存监控装饰器"""
    def __init__(self):
        self.stats = defaultdict(lambda: {'hits': 0, 'misses': 0, 'total_time': 0})
    def monitor_cache(self, cache_name="default"):
        """缓存监控装饰器"""
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                # 模拟缓存key
                cache_key = f"{func.__name__}:{args}:{kwargs}"
                start_time = time.time()
                # 模拟检查缓存
                result = func(*args, **kwargs)
                # 记录统计
                elapsed = time.time() - start_time
                self.stats[cache_name]['total_time'] += elapsed
                return result
            return wrapper
        # 添加统计方法
        decorator.get_stats = lambda: self._get_stats(cache_name)
        return decorator
    def _get_stats(self, cache_name):
        """获取缓存统计"""
        stats = self.stats[cache_name]
        total = stats['hits'] + stats['misses']
        if total > 0:
            hit_rate = (stats['hits'] / total) * 100
        else:
            hit_rate = 0
        return {
            'cache_name': cache_name,
            'hit_rate': round(hit_rate, 2),
            'hits': stats['hits'],
            'misses': stats['misses'],
            'total_requests': total,
            'avg_response_time': round(stats['total_time'] / max(total, 1), 4)
        }
# 使用示例
monitor = CacheMonitor()
@monitor.monitor_cache("user_cache")
def get_user(user_id):
    """模拟获取用户数据"""
    # 模拟缓存命中或未命中
    import random
    if random.random() < 0.7:  # 70%命中率
        return {"user_id": user_id, "name": "User"}
    else:
        return None
# 测试
for _ in range(100):
    get_user(1)
print(monitor.stats)

Nginx 缓存命中率监控

日志分析脚本

#!/bin/bash
# Nginx缓存命中率分析脚本
NGINX_LOG="/var/log/nginx/access.log"
CACHE_HEADER="X-Cache"
analyze_cache_hits() {
    echo "=== Nginx 缓存命中率分析 ==="
    echo "日志文件: $NGINX_LOG"
    echo "------------------------"
    # 统计总请求数
    total=$(wc -l < "$NGINX_LOG")
    # 统计缓存命中（HIT）和未命中（MISS）
    hits=$(grep -c "HIT" "$NGINX_LOG")
    misses=$(grep -c "MISS" "$NGINX_LOG")
    if [ $total -gt 0 ]; then
        hit_rate=$(echo "scale=2; $hits * 100 / $total" | bc)
        echo "总请求数: $total"
        echo "缓存命中: $hits"
        echo "缓存未命中: $misses"
        echo "缓存命中率: ${hit_rate}%"
    fi
    echo "------------------------"
    echo "最近10条请求状态:"
    tail -10 "$NGINX_LOG" | grep -o "$CACHE_HEADER: [A-Z]*" || echo "未找到缓存头信息"
}
# 执行分析
analyze_cache_hits

Prometheus + Grafana 监控

Prometheus 指标导出

from prometheus_client import start_http_server, Gauge, Counter
import redis
import time
import random
class CacheMetrics:
    def __init__(self):
        # Prometheus 指标
        self.cache_hit_rate = Gauge('cache_hit_rate', 'Cache hit rate percentage')
        self.cache_hits = Counter('cache_hits_total', 'Total cache hits')
        self.cache_misses = Counter('cache_misses_total', 'Total cache misses')
        self.redis_client = redis.Redis()
        self.hits_prev = 0
        self.misses_prev = 0
    def collect_metrics(self):
        """收集缓存指标"""
        info = self.redis_client.info('stats')
        hits = info.get('keyspace_hits', 0)
        misses = info.get('keyspace_misses', 0)
        # 更新计数
        hit_diff = hits - self.hits_prev
        miss_diff = misses - self.misses_prev
        for _ in range(hit_diff):
            self.cache_hits.inc()
        for _ in range(miss_diff):
            self.cache_misses.inc()
        # 计算命中率
        total = hits + misses
        if total > 0:
            hit_rate = (hits / total) * 100
            self.cache_hit_rate.set(hit_rate)
        self.hits_prev = hits
        self.misses_prev = misses
if __name__ == '__main__':
    metrics = CacheMetrics()
    start_http_server(8000)
    while True:
        metrics.collect_metrics()
        time.sleep(15)

使用建议

根据缓存类型选择：Redis/Memcached/Nginx各有专门的监控工具
设置告警阈值：一般缓存命中率低于80%需要考虑优化
历史数据存储：建议将监控数据存入时序数据库(InfluxDB/TimescaleDB)
可视化展示：使用Grafana创建缓存监控仪表板
集成到监控系统：配合Prometheus + Alertmanager实现告警

需要我针对特定场景（如Redis集群、多级缓存等）提供更详细的监控方案吗？