实用脚本能自动监控Redis性能吗？

本文目录导读：

1. 轻量级Shell脚本：快速部署，适合小规模环境
2. Python脚本：更强大、更灵活
3. 进阶方案：Prometheus + Grafana
4. 关键性能指标说明
5. 建议

可以的,市面上确实有很多实用的脚本能够自动监控Redis性能，这类脚本通常能帮你收集关键指标，并在出现异常时报警。

下面我为你详细介绍几种常用方案,从轻量级脚本到企业级工具都有，你可以根据实际环境选择。

轻量级Shell脚本：快速部署，适合小规模环境

这种方案依赖Redis自带的redis-cli工具和系统命令，无需额外安装软件。

示例脚本：redis_monitor.sh

#!/bin/bash
# Redis连接信息
REDIS_CLI="/usr/bin/redis-cli"  # 请根据实际路径修改
HOST="127.0.0.1"
PORT="6379"
PASSWORD=""  # 如有密码，填在这里，如 -a 'yourpassword'
# 报警阈值（可根据需要调整）
MAX_MEMORY_MB=1024      # 内存使用超过1024MB报警
MAX_LATENCY_MS=100      # 延迟超过100ms报警
MAX_CONNECTIONS=500     # 连接数超过500报警
REPL_LAG_THRESHOLD=10   # 主从复制延迟超过10秒报警
SLOW_LOG_THRESHOLD=100000  # 慢查询阈值，微秒
# 报警函数（可自定义，如发送邮件、钉钉等）
alert() {
    local message="$1"
    echo "[ALERT] $(date '+%Y-%m-%d %H:%M:%S') - $message" >> /var/log/redis_monitor.log
    # 可在此添加发送邮件或Webhook的代码
    # curl -X POST -H "Content-Type: application/json" -d '{"text":"'"$message"'"}' your_webhook_url
}
# 获取INFO命令的输出
if [ -n "$PASSWORD" ]; then
    INFO=$($REDIS_CLI -h $HOST -p $PORT -a "$PASSWORD" INFO 2>/dev/null)
else
    INFO=$($REDIS_CLI -h $HOST -p $PORT INFO 2>/dev/null)
fi
if [ $? -ne 0 ]; then
    alert "无法连接到Redis服务器 $HOST:$PORT"
    exit 1
fi
# 提取关键性能指标
used_memory=$(echo "$INFO" | grep "^used_memory:" | cut -d ':' -f2)
used_memory_rss=$(echo "$INFO" | grep "^used_memory_rss:" | cut -d ':' -f2)
used_memory_peak=$(echo "$INFO" | grep "^used_memory_peak:" | cut -d ':' -f2)
total_connections_received=$(echo "$INFO" | grep "^total_connections_received:" | cut -d ':' -f2)
connected_clients=$(echo "$INFO" | grep "^connected_clients:" | cut -d ':' -f2)
instantaneous_ops_per_sec=$(echo "$INFO" | grep "^instantaneous_ops_per_sec:" | cut -d ':' -f2)
instantaneous_input_kbps=$(echo "$INFO" | grep "^instantaneous_input_kbps:" | cut -d ':' -f2)
instantaneous_output_kbps=$(echo "$INFO" | grep "^instantaneous_output_kbps:" | cut -d ':' -f2)
rejected_connections=$(echo "$INFO" | grep "^rejected_connections:" | cut -d ':' -f2)
keyspace_hits=$(echo "$INFO" | grep "^keyspace_hits:" | cut -d ':' -f2)
keyspace_misses=$(echo "$INFO" | grep "^keyspace_misses:" | cut -d ':' -f2)
uptime_in_seconds=$(echo "$INFO" | grep "^uptime_in_seconds:" | cut -d ':' -f2)
role=$(echo "$INFO" | grep "^role:" | cut -d ':' -f2)
# 内存检查（转换为MB）
used_memory_mb=$((used_memory / 1024 / 1024))
if [ $used_memory_mb -gt $MAX_MEMORY_MB ]; then
    alert "Redis内存使用超限: ${used_memory_mb}MB (阈值: ${MAX_MEMORY_MB}MB)"
fi
# 连接数检查
if [ "$connected_clients" -gt "$MAX_CONNECTIONS" ]; then
    alert "Redis客户端连接数过高: $connected_clients (阈值: $MAX_CONNECTIONS)"
fi
# 拒绝连接检查（如果有拒绝连接，通常意味着达到最大连接限制）
if [ -n "$rejected_connections" ] && [ "$rejected_connections" -gt 0 ]; then
    alert "Redis出现拒绝连接: $rejected_connections次"
fi
# 主从复制检查（仅对从节点有效）
if [ "$role" = "slave" ]; then
    master_link_status=$(echo "$INFO" | grep "^master_link_status:" | cut -d ':' -f2)
    master_last_io_seconds_ago=$(echo "$INFO" | grep "^master_last_io_seconds_ago:" | cut -d ':' -f2)
    slave_repl_offset=$(echo "$INFO" | grep "^slave_repl_offset:" | cut -d ':' -f2)
    master_repl_offset=$(echo "$INFO" | grep "^master_repl_offset:" | cut -d ':' -f2)
    if [ "$master_link_status" != "up" ]; then
        alert "Redis主从复制链接断开!"
    fi
    if [ -n "$master_last_io_seconds_ago" ] && [ "$master_last_io_seconds_ago" -gt "$REPL_LAG_THRESHOLD" ]; then
        alert "Redis主从复制延迟: ${master_last_io_seconds_ago}秒 (阈值: ${REPL_LAG_THRESHOLD}秒)"
    fi
fi
# 慢查询检查
SLOW_LOG=$($REDIS_CLI -h $HOST -p $PORT SLOWLOG GET 5 2>/dev/null)
if [ -n "$SLOW_LOG" ]; then
    # 简单判断有没有慢查询
    echo "$SLOW_LOG" | while read line; do
        duration=$(echo "$line" | awk '{print $2}' | tr -d ',')  # 微秒
        if [ "$duration" -gt "$SLOW_LOG_THRESHOLD" ]; then
            alert "Redis慢查询记录: 耗时 ${duration}微秒, 命令: $line"
        fi
    done
fi
# 记录正常状态
echo "$(date '+%Y-%m-%d %H:%M:%S') - OK - 内存:${used_memory_mb}MB, 连接:${connected_clients}, OPS:${instantaneous_ops_per_sec}" >> /var/log/redis_monitor.log

使用方法：

1. 保存为 redis_monitor.sh 并赋予执行权限：chmod +x redis_monitor.sh
2. 设置crontab定期执行（例如每1分钟）：* * * * * /path/to/redis_monitor.sh
3. 在脚本中完善 alert() 函数，对接你的报警通道（企业微信、钉钉、邮件等）。

Python脚本：更强大、更灵活

Python方案可以利用redis-py库和丰富的数据处理能力。

示例脚本：redis_perf_monitor.py

#!/usr/bin/env python3
import redis
import time
import smtplib
import json
from datetime import datetime
import os
# Redis连接配置
REDIS_CONFIG = {
    'host': '127.0.0.1',
    'port': 6379,
    'password': None,
    'db': 0,
    'socket_connect_timeout': 5
}
# 报警阈值
ALERTS = {
    'memory_usage_mb': 1024,
    'connected_clients': 500,
    'hit_rate_low': 0.8,  # 缓存命中率低于80%报警
    'latency_ms': 100,
    'rejected_connections': 1,
    'replication_lag_seconds': 10,
}
# 报警配置
ALERT_WEBHOOK_URL = "https://your-alert-webhook-url"  # 如钉钉、企业微信机器人
SMTP_CONFIG = {
    'server': 'smtp.example.com',
    'port': 587,
    'username': 'your@email.com',
    'password': 'your-password',
    'from_addr': 'monitor@example.com',
    'to_addr': 'admin@example.com'
}
class RedisMonitor:
    def __init__(self, redis_config):
        self.client = redis.StrictRedis(**redis_config)
        self.info = {}
        self.alerts = []
    def collect_metrics(self):
        """收集所有关键指标"""
        try:
            self.info = self.client.info()
        except redis.ConnectionError as e:
            self._add_alert(f"无法连接到Redis: {str(e)}")
            return False
        return True
    def check_memory(self):
        """检查内存使用"""
        used_memory = self.info.get('used_memory', 0)
        used_memory_mb = used_memory / 1024 / 1024
        max_memory = self.info.get('maxmemory', 0)
        max_memory_mb = max_memory / 1024 / 1024
        if used_memory_mb > ALERTS['memory_usage_mb']:
            self._add_alert(f"内存使用超限: {used_memory_mb:.1f}MB (阈值: {ALERTS['memory_usage_mb']}MB)")
        if max_memory > 0:
            usage_percent = (used_memory / max_memory) * 100
            if usage_percent > 80:
                self._add_alert(f"内存使用率超过80%: {usage_percent:.1f}%")
        # 检查RSS内存 vs 已使用内存（判断内存碎片）
        used_memory_rss = self.info.get('used_memory_rss', 0)
        if used_memory_rss > 0:
            fragmentation = used_memory_rss / used_memory if used_memory > 0 else 0
            if fragmentation > 1.5:
                self._add_alert(f"内存碎片率高: {fragmentation:.2f}")
    def check_connections(self):
        """检查连接数和拒绝连接"""
        connected_clients = self.info.get('connected_clients', 0)
        rejected_connections = self.info.get('rejected_connections', 0)
        if connected_clients > ALERTS['connected_clients']:
            self._add_alert(f"客户端连接数过高: {connected_clients} (阈值: {ALERTS['connected_clients']})")
        if rejected_connections > ALERTS['rejected_connections']:
            self._add_alert(f"出现拒绝连接: {rejected_connections}次")
    def check_hit_rate(self):
        """检查缓存命中率"""
        hits = self.info.get('keyspace_hits', 0)
        misses = self.info.get('keyspace_misses', 0)
        total = hits + misses
        if total > 0:
            hit_rate = hits / total
            if hit_rate < ALERTS['hit_rate_low']:
                self._add_alert(f"缓存命中率过低: {hit_rate:.1%} (阈值: {ALERTS['hit_rate_low']:.0%})")
    def check_latency(self):
        """检查延迟"""
        try:
            start = time.time()
            self.client.ping()
            latency = (time.time() - start) * 1000
            if latency > ALERTS['latency_ms']:
                self._add_alert(f"Redis延迟过高: {latency:.1f}ms (阈值: {ALERTS['latency_ms']}ms)")
        except Exception as e:
            self._add_alert(f"延迟检查失败: {str(e)}")
    def check_replication(self):
        """检查主从复制状态"""
        role = self.info.get('role', 'master')
        if role == 'slave':
            master_link_status = self.info.get('master_link_status', 'down')
            master_last_io_seconds_ago = self.info.get('master_last_io_seconds_ago', 0)
            if master_link_status != 'up':
                self._add_alert(f"主从复制链接断开! 状态: {master_link_status}")
            if master_last_io_seconds_ago > ALERTS['replication_lag_seconds']:
                self._add_alert(f"主从复制延迟: {master_last_io_seconds_ago}秒 (阈值: {ALERTS['replication_lag_seconds']}秒)")
            # 检查复制积压缓冲区
            repl_backlog_active = self.info.get('repl_backlog_active', 0)
            repl_backlog_size = self.info.get('repl_backlog_size', 0)
            if repl_backlog_active == 0:
                self._add_alert("复制积压缓冲区未启用!")
    def check_slow_log(self):
        """检查慢查询日志"""
        try:
            slow_logs = self.client.slowlog_get(10)
            for log in slow_logs:
                # slow_logs返回的是列表，每个元素是一个元组 (id, timestamp, duration, command, ...)
                if len(log) >= 4:
                    duration = log[2]  # 微秒
                    command = ' '.join(log[3]) if isinstance(log[3], list) else str(log[3])
                    if duration > 100000:  # 100ms以上
                        self._add_alert(f"慢查询: {duration}微秒, 命令: {command}")
        except Exception as e:
            pass
    def _add_alert(self, message):
        """添加报警信息"""
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        alert_msg = f"[{timestamp}] {message}"
        self.alerts.append(alert_msg)
        print(alert_msg)
    def send_alerts(self):
        """发送报警信息（可对接多种通道）"""
        if not self.alerts:
            return
        alert_text = "\n".join(self.alerts)
        # 示例1：发送到Webhook
        self._send_webhook(alert_text)
        # 示例2：发送邮件
        self._send_email("Redis监控报警", alert_text)
    def _send_webhook(self, message):
        """发送Webhook报警"""
        try:
            import requests
            payload = {
                "msgtype": "text",
                "text": {
                    "content": f"# Redis监控报警\n{message}"
                }
            }
            requests.post(ALERT_WEBHOOK_URL, json=payload, timeout=5)
        except Exception as e:
            print(f"发送Webhook报警失败: {e}")
    def _send_email(self, subject, body):
        """发送邮件报警"""
        try:
            with smtplib.SMTP(SMTP_CONFIG['server'], SMTP_CONFIG['port']) as server:
                server.starttls()
                server.login(SMTP_CONFIG['username'], SMTP_CONFIG['password'])
                msg = f"Subject: {subject}\n\n{body}"
                server.sendmail(SMTP_CONFIG['from_addr'], SMTP_CONFIG['to_addr'], msg)
        except Exception as e:
            print(f"发送邮件报警失败: {e}")
    def generate_report(self):
        """生成性能报告"""
        report = {
            "timestamp": datetime.now().isoformat(),
            "host": REDIS_CONFIG['host'],
            "port": REDIS_CONFIG['port'],
            "metrics": {
                "used_memory_mb": self.info.get('used_memory', 0) / 1024 / 1024,
                "connected_clients": self.info.get('connected_clients', 0),
                "ops_per_sec": self.info.get('instantaneous_ops_per_sec', 0),
                "hit_rate": self._calc_hit_rate(),
                "uptime_days": self.info.get('uptime_in_seconds', 0) / 86400,
                "keys_count": sum(self.info.get('db0', {}).get('keys', 0) for key in self.info if key.startswith('db')),
                "avg_ttl": self.client.info('keyspace').get('db0', {}).get('avg_ttl', 0)
            },
            "alerts": self.alerts
        }
        return report
    def _calc_hit_rate(self):
        hits = self.info.get('keyspace_hits', 0)
        misses = self.info.get('keyspace_misses', 0)
        total = hits + misses
        return hits / total if total > 0 else 0
    def run(self):
        """执行监控主流程"""
        if not self.collect_metrics():
            self.send_alerts()
            return
        self.check_memory()
        self.check_connections()
        self.check_hit_rate()
        self.check_latency()
        self.check_replication()
        self.check_slow_log()
        if self.alerts:
            self.send_alerts()
            self._log_to_file(self.generate_report())
        else:
            print(f"[{datetime.now()}] 性能正常")
    def _log_to_file(self, report):
        """将报告写入日志文件"""
        log_dir = "/var/log/redis_monitor"
        os.makedirs(log_dir, exist_ok=True)
        log_file = f"{log_dir}/redis_monitor_{datetime.now().strftime('%Y%m%d')}.log"
        with open(log_file, 'a') as f:
            f.write(json.dumps(report) + "\n")
if __name__ == "__main__":
    monitor = RedisMonitor(REDIS_CONFIG)
    monitor.run()

进阶方案：Prometheus + Grafana

对于生产环境,推荐使用成熟的监控体系：

1. 数据采集：使用 redis_exporter（一个专门导出Redis指标的Prometheus exporter）
2. 数据存储：Prometheus负责抓取和存储指标
3. 可视化：Grafana提供丰富的仪表盘

快速部署命令：

# 启动Redis Exporter
docker run -d --name redis_exporter \
  -e REDIS_ADDR=redis://your-redis-host:6379 \
  -p 9121:9121 \
  oliver006/redis_exporter
# 配置Prometheus（prometheus.yml）
# 添加：
# - job_name: 'redis'
#   static_configs:
#     - targets: ['localhost:9121']
# Grafana导入Redis仪表盘（ID: 11835 或 763）

关键性能指标说明

无论使用哪种方案,都应关注以下核心指标：

建议

• 脚本化监控适合快速部署，但功能有限
• Prometheus+Grafana是生产环境首选
• 对于云Redis（如阿里云、腾讯云），建议使用云厂商自带的监控服务
• 报警规则要合理：避免阈值过低导致报警风暴，也要避免过高导致故障遗漏
• 结合日志分析：将监控数据与业务日志关联，定位根因

如果需要更详细的配置说明或针对特定场景（如集群模式、哨兵模式）的监控脚本，可以进一步沟通。