本文目录导读：
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
这些脚本提供了完整的缓存监控解决方案,包括实时监控、性能测试、告警机制等，可以根据实际需求进行定制和扩展。
