怎样实现缓存监控脚本

wen 实用脚本 2026-07-01 201

本文目录导读：

怎样实现缓存监控脚本

Redis 缓存监控脚本
Memcached 监控脚本
通用缓存状态检查脚本
缓存性能测试脚本
使用说明

Redis 缓存监控脚本

#!/usr/bin/env python3
# redis_monitor.py - Redis缓存监控脚本
import redis
import json
import time
from datetime import datetime
import argparse
class RedisCacheMonitor:
    def __init__(self, host='localhost', port=6379, password=None, db=0):
        self.redis_client = redis.Redis(
            host=host,
            port=port,
            password=password,
            db=db,
            decode_responses=True
        )
    def get_basic_stats(self):
        """获取基本统计信息"""
        info = self.redis_client.info()
        return {
            'connected_clients': info['connected_clients'],
            'used_memory': info['used_memory_human'],
            'used_memory_peak': info['used_memory_peak_human'],
            'total_commands_processed': info['total_commands_processed'],
            'keyspace_hits': info['keyspace_hits'],
            'keyspace_misses': info['keyspace_misses'],
            'hit_rate': self._calculate_hit_rate(info),
            'uptime_in_days': info['uptime_in_days']
        }
    def _calculate_hit_rate(self, info):
        """计算缓存命中率"""
        hits = info['keyspace_hits']
        misses = info['keyspace_misses']
        total = hits + misses
        if total == 0:
            return 0
        return round((hits / total) * 100, 2)
    def get_keyspace_stats(self):
        """获取键空间统计"""
        db_stats = self.redis_client.info('keyspace')
        stats = {}
        for db, info in db_stats.items():
            parts = info.split(',')
            keys = int(parts[0].split('=')[1])
            expires = int(parts[1].split('=')[1]) if len(parts) > 1 else 0
            stats[db] = {
                'keys': keys,
                'expires': expires,
                'avg_ttl': parts[2].split('=')[1] if len(parts) > 2 else 'N/A'
            }
        return stats
    def get_slow_queries(self, threshold=1000):
        """获取慢查询日志"""
        slow_logs = self.redis_client.slowlog_get()
        slow_queries = []
        for log in slow_logs:
            if log['duration'] > threshold:
                slow_queries.append({
                    'id': log['id'],
                    'duration': log['duration'],
                    'command': ' '.join(log['args']),
                    'timestamp': datetime.fromtimestamp(log['start_time']).isoformat()
                })
        return slow_queries
    def check_replication(self):
        """检查复制状态"""
        info = self.redis_client.info('replication')
        return {
            'role': info['role'],
            'connected_slaves': info['connected_slaves'],
            'master_link_status': info.get('master_link_status', 'N/A'),
            'master_last_io_seconds_ago': info.get('master_last_io_seconds_ago', 0)
        }
    def monitor_cpu(self):
        """监控CPU使用情况"""
        info = self.redis_client.info('cpu')
        return {
            'used_cpu_sys': info['used_cpu_sys'],
            'used_cpu_user': info['used_cpu_user'],
            'used_cpu_sys_children': info['used_cpu_sys_children'],
            'used_cpu_user_children': info['used_cpu_user_children']
        }
def main():
    parser = argparse.ArgumentParser(description='Redis缓存监控')
    parser.add_argument('--host', default='localhost', help='Redis主机')
    parser.add_argument('--port', type=int, default=6379, help='Redis端口')
    parser.add_argument('--password', help='Redis密码')
    parser.add_argument('--alert-threshold', type=float, default=80, 
                       help='内存使用告警阈值（百分比）')
    parser.add_argument('--interval', type=int, default=10, 
                       help='监控间隔（秒）')
    args = parser.parse_args()
    monitor = RedisCacheMonitor(args.host, args.port, args.password)
    print(f"开始监控Redis缓存 ({args.host}:{args.port})")
    print("=" * 50)
    try:
        while True:
            print(f"\n[{datetime.now().isoformat()}]")
            # 基本统计
            stats = monitor.get_basic_stats()
            print(f"客户端连接数: {stats['connected_clients']}")
            print(f"内存使用: {stats['used_memory']}")
            print(f"峰值内存: {stats['used_memory_peak']}")
            print(f"缓存命中率: {stats['hit_rate']}%")
            # 内存告警
            used_memory_bytes = int(stats['used_memory'].replace('M', '000000').replace('K', '000'))
            total_memory = 512 * 1024 * 1024  # 假设512MB
            if used_memory_bytes > total_memory * args.alert_threshold / 100:
                print(f"⚠️  警告：内存使用超过{args.alert_threshold}%!")
            # 慢查询
            slow_queries = monitor.get_slow_queries()
            if slow_queries:
                print(f"慢查询数: {len(slow_queries)}")
            time.sleep(args.interval)
    except KeyboardInterrupt:
        print("\n监控已停止")
    except redis.exceptions.ConnectionError:
        print("无法连接到Redis服务器")
if __name__ == "__main__":
    main()

Memcached 监控脚本

#!/usr/bin/env python3
# memcached_monitor.py
import telnetlib
import time
from datetime import datetime
class MemcachedMonitor:
    def __init__(self, host='localhost', port=11211, timeout=5):
        self.host = host
        self.port = port
        self.timeout = timeout
    def _execute_command(self, command):
        """执行telnet命令"""
        try:
            tn = telnetlib.Telnet(self.host, self.port, self.timeout)
            tn.write(f"{command}\r\n".encode())
            tn.write(b"quit\r\n")
            output = tn.read_all().decode()
            tn.close()
            return output
        except Exception as e:
            return f"Error: {str(e)}"
    def get_stats(self):
        """获取统计信息"""
        output = self._execute_command("stats")
        stats = {}
        for line in output.split('\n'):
            if line.startswith('STAT'):
                parts = line.split()
                if len(parts) >= 3:
                    key = parts[1]
                    value = parts[2]
                    stats[key] = value
        return stats
    def calculate_metrics(self, stats):
        """计算关键指标"""
        metrics = {}
        # 当前连接数
        metrics['curr_connections'] = int(stats.get('curr_connections', 0))
        # 缓存命中率
        get_hits = int(stats.get('get_hits', 0))
        get_misses = int(stats.get('get_misses', 0))
        total_gets = get_hits + get_misses
        metrics['hit_rate'] = (get_hits / total_gets * 100) if total_gets > 0 else 0
        # 内存使用率
        curr_bytes = int(stats.get('bytes', 0))
        limit_maxbytes = int(stats.get('limit_maxbytes', 0))
        metrics['memory_usage'] = (curr_bytes / limit_maxbytes * 100) if limit_maxbytes > 0 else 0
        # 逐出率
        evictions = int(stats.get('evictions', 0))
        cmd_set = int(stats.get('cmd_set', 0))
        metrics['eviction_rate'] = (evictions / cmd_set * 100) if cmd_set > 0 else 0
        # 当前项目数
        metrics['current_items'] = int(stats.get('curr_items', 0))
        return metrics
    def check_health(self, metrics):
        """检查健康状态"""
        issues = []
        if metrics['hit_rate'] < 80:
            issues.append(f"缓存命中率低: {metrics['hit_rate']:.2f}%")
        if metrics['memory_usage'] > 90:
            issues.append(f"内存使用率高: {metrics['memory_usage']:.2f}%")
        if metrics['eviction_rate'] > 5:
            issues.append(f"数据逐出率高: {metrics['eviction_rate']:.2f}%")
        return issues
def main():
    monitor = MemcachedMonitor()
    print("Memcached 缓存监控")
    print("=" * 50)
    while True:
        try:
            print(f"\n[{datetime.now().isoformat()}]")
            stats = monitor.get_stats()
            metrics = monitor.calculate_metrics(stats)
            print(f"当前连接数: {metrics['curr_connections']}")
            print(f"当前项目数: {metrics['current_items']}")
            print(f"缓存命中率: {metrics['hit_rate']:.2f}%")
            print(f"内存使用率: {metrics['memory_usage']:.2f}%")
            print(f"数据逐出率: {metrics['eviction_rate']:.2f}%")
            # 健康检查
            issues = monitor.check_health(metrics)
            if issues:
                print("\n⚠️  告警:")
                for issue in issues:
                    print(f"  - {issue}")
            time.sleep(10)
        except KeyboardInterrupt:
            print("\n监控已停止")
            break
        except Exception as e:
            print(f"错误: {e}")
            time.sleep(5)
if __name__ == "__main__":
    main()

通用缓存状态检查脚本

#!/bin/bash
# cache_check.sh - 通用缓存状态检查
# 配置
REDIS_HOST="localhost"
REDIS_PORT=6379
MEMCACHED_HOST="localhost"
MEMCACHED_PORT=11211
LOG_FILE="/var/log/cache_monitor.log"
ALERT_EMAIL="admin@example.com"
# 日志函数
log_message() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}
# Redis检查函数
check_redis() {
    log_message "检查Redis缓存..."
    # 检查Redis是否运行
    if command -v redis-cli &> /dev/null; then
        REDIS_STATUS=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" ping 2>/dev/null)
        if [ "$REDIS_STATUS" = "PONG" ]; then
            log_message "Redis状态: 运行中"
            # 获取关键指标
            MEMORY_USAGE=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" INFO memory | grep "used_memory_human" | cut -d: -f2)
            HIT_RATE=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" INFO stats | grep "keyspace_hits" | cut -d: -f2)
            MISS_RATE=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" INFO stats | grep "keyspace_misses" | cut -d: -f2)
            log_message "内存使用: ${MEMORY_USAGE:-未知}"
            # 计算命中率
            if [ -n "$HIT_RATE" ] && [ -n "$MISS_RATE" ]; then
                TOTAL=$((HIT_RATE + MISS_RATE))
                if [ $TOTAL -gt 0 ]; then
                    HIT_PERCENT=$(echo "scale=2; $HIT_RATE * 100 / $TOTAL" | bc)
                    log_message "缓存命中率: ${HIT_PERCENT}%"
                    # 告警阈值检查
                    if (( $(echo "$HIT_PERCENT < 80" | bc -l) )); then
                        log_message "警告: 缓存命中率低于80%"
                        # 发送告警邮件
                        echo "Redis缓存命中率低于80%，当前: ${HIT_PERCENT}%" | \
                            mail -s "缓存告警" "$ALERT_EMAIL"
                    fi
                fi
            fi
        else
            log_message "错误: Redis未响应"
        fi
    else
        log_message "错误: redis-cli命令未找到"
    fi
}
# Memcached检查函数
check_memcached() {
    log_message "检查Memcached缓存..."
    # 使用echo和nc检查memcached
    if command -v nc &> /dev/null; then
        MEMCACHED_STATS=$(echo -e "stats\nquit" | nc -w 2 "$MEMCACHED_HOST" "$MEMCACHED_PORT" 2>/dev/null)
        if [ -n "$MEMCACHED_STATS" ]; then
            log_message "Memcached状态: 运行中"
            # 解析统计信息
            CURR_CONNECTIONS=$(echo "$MEMCACHED_STATS" | grep "curr_connections" | awk '{print $3}')
            GET_HITS=$(echo "$MEMCACHED_STATS" | grep "get_hits" | awk '{print $3}')
            GET_MISSES=$(echo "$MEMCACHED_STATS" | grep "get_misses" | awk '{print $3}')
            CURR_ITEMS=$(echo "$MEMCACHED_STATS" | grep "curr_items" | awk '{print $3}')
            log_message "当前连接数: ${CURR_CONNECTIONS:-未知}"
            log_message "当前项目数: ${CURR_ITEMS:-未知}"
            # 计算命中率
            if [ -n "$GET_HITS" ] && [ -n "$GET_MISSES" ]; then
                TOTAL_GETS=$((GET_HITS + GET_MISSES))
                if [ $TOTAL_GETS -gt 0 ]; then
                    HIT_PERCENT=$(echo "scale=2; $GET_HITS * 100 / $TOTAL_GETS" | bc)
                    log_message "缓存命中率: ${HIT_PERCENT}%"
                fi
            fi
        else
            log_message "错误: Memcached未响应"
        fi
    else
        log_message "错误: nc命令未找到"
    fi
}
# 主函数
main() {
    log_message "===== 缓存监控检查开始 ====="
    # 检查所有缓存服务
    check_redis
    check_memcached
    # 检查系统内存
    log_message "系统内存使用情况:"
    free -h | grep -E "Mem:|Swap:" | while read line; do
        log_message "  $line"
    done
    log_message "===== 缓存监控检查完成 ====="
    echo ""
}
# 执行主函数
main

缓存性能测试脚本

#!/usr/bin/env python3
# cache_benchmark.py - 缓存性能测试
import time
import random
import string
import redis
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
class CacheBenchmark:
    def __init__(self, host='localhost', port=6379):
        self.client = redis.Redis(host=host, port=port, decode_responses=True)
    def generate_data(self, size=1000):
        """生成测试数据"""
        data = {}
        for i in range(size):
            key = f"benchmark:{i}"
            value = ''.join(random.choices(string.ascii_letters + string.digits, k=100))
            data[key] = value
        return data
    def test_write_performance(self, data, threads=10):
        """测试写入性能"""
        items = list(data.items())
        chunk_size = len(items) // threads
        def write_chunk(chunk):
            start = time.time()
            for key, value in chunk:
                self.client.set(key, value, ex=300)  # 5分钟过期
            return time.time() - start
        with ThreadPoolExecutor(max_workers=threads) as executor:
            chunks = [items[i:i+chunk_size] for i in range(0, len(items), chunk_size)]
            futures = [executor.submit(write_chunk, chunk) for chunk in chunks]
            total_time = 0
            for future in as_completed(futures):
                total_time += future.result()
            ops_per_sec = len(items) / total_time
            return {
                'total_ops': len(items),
                'total_time': round(total_time, 3),
                'ops_per_sec': round(ops_per_sec, 2),
                'avg_latency': round(total_time / len(items) * 1000, 3)  # ms
            }
    def test_read_performance(self, keys, threads=10):
        """测试读取性能"""
        def read_keys(key_list):
            start = time.time()
            for key in key_list:
                self.client.get(key)
            return time.time() - start
        chunk_size = len(keys) // threads
        with ThreadPoolExecutor(max_workers=threads) as executor:
            chunks = [keys[i:i+chunk_size] for i in range(0, len(keys), chunk_size)]
            futures = [executor.submit(read_keys, chunk) for chunk in chunks]
            total_time = 0
            for future in as_completed(futures):
                total_time += future.result()
            ops_per_sec = len(keys) / total_time
            return {
                'total_ops': len(keys),
                'total_time': round(total_time, 3),
                'ops_per_sec': round(ops_per_sec, 2),
                'avg_latency': round(total_time / len(keys) * 1000, 3)
            }
    def cleanup(self):
        """清理测试数据"""
        for key in self.client.scan_iter("benchmark:*"):
            self.client.delete(key)
def main():
    benchmark = CacheBenchmark()
    data_size = int(sys.argv[1]) if len(sys.argv) > 1 else 1000
    print("缓存性能测试")
    print("=" * 50)
    try:
        # 清理旧数据
        benchmark.cleanup()
        # 生成测试数据
        print(f"生成{data_size}条测试数据...")
        data = benchmark.generate_data(data_size)
        # 写入测试
        print("\n写入性能测试:")
        write_result = benchmark.test_write_performance(data)
        print(f"  总操作数: {write_result['total_ops']}")
        print(f"  总耗时: {write_result['total_time']}秒")
        print(f"  每秒操作数: {write_result['ops_per_sec']}")
        print(f"  平均延迟: {write_result['avg_latency']}ms")
        # 读取测试
        print("\n读取性能测试:")
        keys = list(data.keys())
        read_result = benchmark.test_read_performance(keys)
        print(f"  总操作数: {read_result['total_ops']}")
        print(f"  总耗时: {read_result['total_time']}秒")
        print(f"  每秒操作数: {read_result['ops_per_sec']}")
        print(f"  平均延迟: {read_result['avg_latency']}ms")
    finally:
        # 清理测试数据
        benchmark.cleanup()
        print("\n测试完成，已清理测试数据")
if __name__ == "__main__":
    main()

使用说明

安装依赖

# Redis监控
pip install redis
# Memcached监控
# 使用系统自带的telnet/nc即可
# 性能测试
pip install redis

运行脚本

# Redis监控
python redis_monitor.py --host localhost --port 6379
# Memcached监控
python memcached_monitor.py
# 缓存检查脚本
chmod +x cache_check.sh
./cache_check.sh
# 性能测试
python cache_benchmark.py 1000

配合crontab定期执行

# 每5分钟执行一次缓存检查
*/5 * * * * /path/to/cache_check.sh
# 每天凌晨执行性能测试
0 2 * * * /usr/bin/python3 /path/to/cache_benchmark.py 10000 >> /var/log/cache_benchmark.log

这些脚本提供了完整的缓存监控解决方案,包括实时监控、性能测试、告警机制等，可以根据实际需求进行定制和扩展。

本文地址： https://www.bmjup.com/post/13031.html