如何用脚本分析Redis慢查询？

wen 实用脚本 2026-07-21 166

本文目录导读：

如何用脚本分析Redis慢查询？

基础慢查询分析脚本
高级分析脚本
使用方式

我来介绍几种分析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

这些脚本可以帮你：

实时监控慢查询
统计分析慢查询模式
生成报告便于分析
提供优化建议

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