本文目录导读：
我来介绍几种分析Redis慢查询的脚本方法：
基础慢查询分析脚本
Python脚本
#!/usr/bin/env python3
import redis
import time
from datetime import datetime
import argparse
class RedisSlowLogAnalyzer:
def __init__(self, host='localhost', port=6379, password=None):
self.redis_client = redis.Redis(
host=host,
port=port,
password=password,
decode_responses=True
)
def get_slow_logs(self, length=100):
"""获取慢查询日志"""
try:
logs = self.redis_client.slowlog_get(length)
return logs
except Exception as e:
print(f"获取慢查询日志失败: {e}")
return []
def analyze_slow_logs(self, logs):
"""分析慢查询日志"""
if not logs:
print("没有慢查询记录")
return
print(f"\n=== Redis慢查询分析报告 ({datetime.now()}) ===")
print(f"共发现 {len(logs)} 条慢查询记录\n")
# 统计信息
total_duration = 0
command_stats = {}
for log in logs:
log_id = log['id']
timestamp = log['start_time']
duration = log['duration']
command = ' '.join(log['command'])
# 耗时转换为毫秒
duration_ms = duration / 1000
total_duration += duration_ms
# 统计命令类型
cmd_type = log['command'][0] if log['command'] else 'unknown'
if cmd_type not in command_stats:
command_stats[cmd_type] = {
'count': 0,
'total_duration': 0,
'max_duration': 0,
'commands': []
}
command_stats[cmd_type]['count'] += 1
command_stats[cmd_type]['total_duration'] += duration_ms
command_stats[cmd_type]['max_duration'] = max(
command_stats[cmd_type]['max_duration'],
duration_ms
)
command_stats[cmd_type]['commands'].append({
'id': log_id,
'time': datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S'),
'duration_ms': round(duration_ms, 2),
'command': command
})
# 打印详细记录
print(f"[{log_id}] 时间: {datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')}")
print(f" 耗时: {round(duration_ms, 2)}ms")
print(f" 命令: {command}")
print()
# 打印统计信息
print("=== 按命令类型统计 ===")
for cmd_type, stats in sorted(
command_stats.items(),
key=lambda x: x[1]['total_duration'],
reverse=True
):
avg_duration = stats['total_duration'] / stats['count']
print(f"{cmd_type}: {stats['count']}次, "
f"总耗时: {round(stats['total_duration'], 2)}ms, "
f"平均: {round(avg_duration, 2)}ms, "
f"最大: {round(stats['max_duration'], 2)}ms")
print(f"\n总耗时: {round(total_duration, 2)}ms")
print(f"平均耗时: {round(total_duration/len(logs), 2)}ms")
def configure_slow_log(redis_client, threshold=10000, maxlen=128):
"""配置慢查询参数"""
# 设置慢查询阈值（微秒），默认10ms
redis_client.config_set('slowlog-log-slower-than', threshold)
# 设置最大记录数
redis_client.config_set('slowlog-max-len', maxlen)
print(f"慢查询配置已更新: 阈值={threshold}μs, 最大记录数={maxlen}")
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('--count', type=int, default=100, help='获取记录数')
parser.add_argument('--config', action='store_true', help='配置慢查询参数')
parser.add_argument('--threshold', type=int, default=10000,
help='慢查询阈值（微秒）')
parser.add_argument('--maxlen', type=int, default=128,
help='最大记录数')
args = parser.parse_args()
analyzer = RedisSlowLogAnalyzer(args.host, args.port, args.password)
if args.config:
configure_slow_log(analyzer.redis_client, args.threshold, args.maxlen)
logs = analyzer.get_slow_logs(args.count)
analyzer.analyze_slow_logs(logs)
if __name__ == '__main__':
main()
Shell脚本
#!/bin/bash
# Redis慢查询分析脚本
REDIS_CLI="redis-cli"
REDIS_HOST="localhost"
REDIS_PORT=6379
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'
echo -e "${GREEN}=== Redis慢查询分析工具 ===${NC}\n"
# 检查Redis连接
if ! $REDIS_CLI -h $REDIS_HOST -p $REDIS_PING > /dev/null 2>&1; then
echo -e "${RED}错误: 无法连接到Redis服务器${NC}"
exit 1
fi
# 获取当前慢查询配置
echo -e "${YELLOW}当前慢查询配置:${NC}"
$REDIS_CLI -h $REDIS_HOST -p $REDIS_PORT CONFIG GET slowlog-log-slower-than
$REDIS_CLI -h $REDIS_HOST -p $REDIS_PORT CONFIG GET slowlog-max-len
echo ""
# 获取慢查询日志长度
LEN=$($REDIS_CLI -h $REDIS_HOST -p $REDIS_PORT SLOWLOG LEN)
echo -e "${YELLOW}当前慢查询记录数: ${LEN}${NC}\n"
# 获取并分析慢查询日志
echo -e "${YELLOW}最近10条慢查询记录:${NC}"
$REDIS_CLI -h $REDIS_HOST -p $REDIS_PORT --raw SLOWLOG GET 10 | while read line; do
case $1 in
"1") echo -e "${RED}ID: $line${NC}" ;;
"2") echo -e "时间戳: $(date -d @$line '+%Y-%m-%d %H:%M:%S')" ;;
"3") echo -e "耗时: $(($line / 1000))ms" ;;
"4") echo -e "命令: $line" ;;
*) echo "" ;;
esac
done
# 生成统计报告
echo -e "\n${GREEN}=== 命令类型统计 ===${NC}"
$REDIS_CLI -h $REDIS_HOST -p $REDIS_PORT SLOWLOG GET 100 | \
awk '
BEGIN { FS="\n"; RS="" }
{
for(i=1; i<=NF; i++) {
if($i ~ /^4\)/) {
cmd = $i
gsub(/^4\) "/, "", cmd)
gsub(/".*$/, "", cmd)
split(cmd, parts, " ")
cmds[parts[1]]++
total[parts[1]]++
}
if($i ~ /^3\)/) {
dur = $i
gsub(/^3\) /, "", dur)
total_duration += dur
if(dur > max_duration) max_duration = dur
}
}
}
END {
print "命令类型 | 次数 | 占比"
print "---------|------|------"
for(cmd in cmds) {
printf "%-10s | %-4d | %.1f%%\n", cmd, cmds[cmd], (cmds[cmd]/total)*100
}
}'
# 清理慢查询日志（可选）
echo -e "\n${YELLOW}是否清理慢查询日志? (y/n)${NC}"
read answer
if [ "$answer" = "y" ]; then
$REDIS_CLI -h $REDIS_HOST -p $REDIS_PORT SLOWLOG RESET
echo -e "${GREEN}慢查询日志已清理${NC}"
fi
高级分析脚本
Python持续监控脚本
#!/usr/bin/env python3
import redis
import time
import json
from datetime import datetime
import signal
import sys
class RedisSlowLogMonitor:
def __init__(self, host='localhost', port=6379, password=None):
self.redis_client = redis.Redis(
host=host, port=port, password=password,
decode_responses=True
)
self.running = True
self.last_log_id = 0
def analyze_command_patterns(self, logs):
"""分析慢查询模式"""
patterns = {
'keys_scan': {'pattern': ['keys', 'scan'], 'count': 0},
'complex_commands': {'pattern': ['sort', 'lrange', 'zrange'], 'count': 0},
'big_values': {'pattern': ['get', 'hgetall', 'smembers'], 'count': 0},
'transactions': {'pattern': ['multi', 'exec', 'watch'], 'count': 0}
}
for log in logs:
command = ' '.join(log['command']).lower()
for pattern_name, pattern_info in patterns.items():
for p in pattern_info['pattern']:
if p in command:
pattern_info['count'] += 1
break
return patterns
def suggest_optimization(self, command, duration_ms):
"""提供优化建议"""
suggestions = []
cmd = command[0].upper() if command else ''
if cmd == 'KEYS':
suggestions.append("使用SCAN替代KEYS，避免阻塞")
elif cmd == 'SORT':
suggestions.append("考虑使用有序集合或添加索引")
elif duration_ms > 100:
suggestions.append(f"命令耗时{round(duration_ms,2)}ms，建议优化查询或添加缓存")
if len(command) > 10:
suggestions.append("命令参数过多，建议分批执行")
return suggestions
def monitor_loop(self, interval=5, max_logs=50):
"""持续监控循环"""
print(f"开始监控Redis慢查询 (间隔: {interval}秒)")
print("按 Ctrl+C 停止监控\n")
while self.running:
try:
logs = self.redis_client.slowlog_get(max_logs)
if logs:
new_logs = [log for log in logs if log['id'] > self.last_log_id]
if new_logs:
current_time = datetime.now().strftime('%H:%M:%S')
print(f"\n[{current_time}] 发现 {len(new_logs)} 条新慢查询:")
for log in new_logs:
duration_ms = log['duration'] / 1000
command = ' '.join(log['command'])
print(f" ├ ID: {log['id']}")
print(f" ├ 耗时: {round(duration_ms, 2)}ms")
print(f" ├ 命令: {command}")
suggestions = self.suggest_optimization(
log['command'], duration_ms
)
if suggestions:
print(f" └ 建议: {', '.join(suggestions)}")
print()
self.last_log_id = max(log['id'] for log in logs)
# 分析命令模式
patterns = self.analyze_command_patterns(logs)
print("慢查询模式分析:")
for pattern_name, pattern_info in patterns.items():
if pattern_info['count'] > 0:
print(f" ├ {pattern_name}: {pattern_info['count']}次")
print()
time.sleep(interval)
except KeyboardInterrupt:
self.stop()
except Exception as e:
print(f"错误: {e}")
time.sleep(interval)
def stop(self):
"""停止监控"""
self.running = False
print("\n监控已停止")
def generate_report(logs, filename='slowlog_report.json'):
"""生成JSON格式报告"""
report = {
'timestamp': datetime.now().isoformat(),
'total_logs': len(logs),
'logs': []
}
for log in logs:
log_entry = {
'id': log['id'],
'timestamp': datetime.fromtimestamp(log['start_time']).isoformat(),
'duration_ms': log['duration'] / 1000,
'command': ' '.join(log['command']),
'command_type': log['command'][0] if log['command'] else 'unknown'
}
report['logs'].append(log_entry)
with open(filename, 'w') as f:
json.dump(report, f, indent=2)
print(f"报告已生成: {filename}")
# 使用示例
if __name__ == '__main__':
monitor = RedisSlowLogMonitor()
# 获取当前慢查询并生成报告
logs = monitor.redis_client.slowlog_get(100)
generate_report(logs)
# 启动实时监控
monitor.monitor_loop()
使用方式
# Python脚本使用 python3 redis_slowlog_analyzer.py --host localhost --port 6379 python3 redis_slowlog_analyzer.py --config --threshold 5000 --maxlen 200 # Shell脚本使用 chmod +x redis_slowlog_analyzer.sh ./redis_slowlog_analyzer.sh # 持续监控 python3 redis_slowlog_monitor.py
这些脚本可以帮你：
- 实时监控慢查询
- 统计分析慢查询模式
- 生成报告便于分析
- 提供优化建议
