4. 缓存命中率数据采集脚本编写：Python/Go脚本实现每分钟采集、写入时序数据库、异常告警

4.1 采集原理与设计目标

缓存命中率是衡量缓存系统健康度的核心指标，需要以分钟级粒度持续采集。手动检查不现实，必须由自动化脚本完成：每隔60秒连接缓存实例（如 Redis、Memcached），解析 INFO 或 stats 命令返回的命中次数与总请求次数，计算实时命中率，然后将其写入时序数据库，同时将低于阈值的异常数据推送到告警通道。

Python 与 Go 是该场景最常用的两种语言：Python 开发快、生态丰富，Go 部署简单、并发性能高。本章同时给出两种实现，并对比其差异。

4.2 采集架构概览

4.3 Python 采集脚本实现

4.3.1 依赖安装

pip install redis influxdb-client requests

4.3.2 脚本核心代码

以下脚本每分钟采集一次 Redis 命中率，写入 InfluxDB，并在命中率低于 0.9 时通过 HTTP Webhook 告警。

import time
import redis
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import requests

# === 配置 ===
REDIS_HOST = '192.168.1.100'
REDIS_PORT = 6379
INFLUX_URL = 'http://192.168.1.200:8086'
INFLUX_ = 'your-'
INFLUX_ORG = 'monitor'
INFLUX_BUCKET = 'cache_metrics'
ALERT_URL = 'http://alert.example.com/webhook'
ALERT_THRESHOLD = 0.9

def get_redis_hit_rate(client):
    """从 Redis INFO 中解析命中率"""
    info = client.info('stats')
    hits = info.get('keyspace_hits', 0)
    misses = info.get('keyspace_misses', 0)
    total = hits + misses
    if total == 0:
        return 0.0
    return hits / total

def write_to_influx(point):
    """写入时序数据库"""
    with InfluxDBClient(url=INFLUX_URL, =INFLUX_, org=INFLUX_ORG) as client:
        write_api = client.write_api(write_options=SYNCHRONOUS)
        write_api.write(bucket=INFLUX_BUCKET, record=point)

def send_alert(rate):
    """触发告警"""
    payload = {
        'metric': 'cache_hit_rate',
        'value': round(rate, 4),
        'threshold': ALERT_THRESHOLD,
        'message': f'缓存命中率 {rate:.2%} 低于阈值 {ALERT_THRESHOLD:.0%}'
    }
    try:
        requests.post(ALERT_URL, json=payload, timeout=5)
    except Exception as e:
        print(f'告警发送失败: {e}')

def collect_and_report():
    try:
        r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, socket_connect_timeout=3)
        rate = get_redis_hit_rate(r)
    except redis.ConnectionError:
        # 连接异常时写入 NaN 并告警
        rate = -1.0
        send_alert(-1.0)  # 特殊值代表不可用
        return

    point = Point('cache_hit_rate') \
        .tag('host', REDIS_HOST) \
        .field('value', round(rate, 4)) \
        .time(time.time_ns(), write_precision='ns')
    write_to_influx(point)

    if rate < ALERT_THRESHOLD and rate >= 0:
        send_alert(rate)

if __name__ == '__main__':
    # 实际生产中由 cron/systemd timer 每分钟调用
    collect_and_report()

4.3.3 定时执行（Linux crontab）

# 每分钟执行一次
* * * * * /usr/bin/python3 /opt/scripts/collect_hit_rate.py

4.4 Go 采集脚本实现

4.4.1 依赖管理

go get github.com/go-redis/redis/v8
go get github.com/influxdata/influxdb-client-go/v2

4.4.2 脚本核心代码

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"

    "github.com/go-redis/redis/v8"
    influxdb2 "github.com/influxdata/influxdb-client-go/v2"
)

var (
    redisHost      = "192.168.1.100:6379"
    influxURL      = "http://192.168.1.200:8086"
    influxToken    = "your-"
    influxOrg      = "monitor"
    influxBucket   = "cache_metrics"
    alertURL       = "http://alert.example.com/webhook"
    alertThreshold = 0.9
)

func getHitRate(ctx context.Context, client *redis.Client) (float64, error) {
    info, err := client.Info(ctx, "stats").Result()
    if err != nil {
        return 0, err
    }
    // 解析 INFO 文本
    hits := 0.0
    misses := 0.0
    for _, line := range strings.Split(info, "\n") {
        if strings.HasPrefix(line, "keyspace_hits:") {
            v := strings.TrimPrefix(line, "keyspace_hits:")
            hits, _ = strconv.ParseFloat(strings.TrimSpace(v), 64)
        } else if strings.HasPrefix(line, "keyspace_misses:") {
            v := strings.TrimPrefix(line, "keyspace_misses:")
            misses, _ = strconv.ParseFloat(strings.TrimSpace(v), 64)
        }
    }
    total := hits + misses
    if total == 0 {
        return 0, nil
    }
    return hits / total, nil
}

func writePoint(rate float64) {
    client := influxdb2.NewClient(influxURL, influxToken)
    defer client.Close()
    writeAPI := client.WriteAPIBlocking(influxOrg, influxBucket)
    p := influxdb2.NewPoint("cache_hit_rate",
        map[string]string{"host": redisHost},
        map[string]interface{}{"value": rate},
        time.Now())
    if err := writeAPI.WritePoint(context.Background(), p); err != nil {
        log.Printf("写入 InfluxDB 失败: %v", err)
    }
}

func sendAlert(rate float64) {
    msg := fmt.Sprintf(`{"metric":"cache_hit_rate","value":%.4f,"threshold":%.2f,"message":"命中率 %.2f%% 低于阈值"}`, rate, alertThreshold, rate*100)
    http.Post(alertURL, "application/json", strings.NewReader(msg))
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr:        redisHost,
        DialTimeout: 3 * time.Second,
    })
    defer rdb.Close()

    rate, err := getHitRate(ctx, rdb)
    if err != nil {
        log.Printf("Redis 连接或查询失败: %v", err)
        rate = -1.0
        sendAlert(-1.0) // 不可达告警
        return
    }

    writePoint(rate)
    log.Printf("采集命中率: %.4f", rate)

    if rate >= 0 && rate < alertThreshold {
        sendAlert(rate)
    }
}

4.4.3 编译与部署

# 编译为可独立运行的二进制
go build -o collect_hit_rate collect_hit_rate.go
# 使用 systemd timer 每分钟执行
# 创建 unit 文件和 timer 文件

4.5 脚本对比

4.6 调试与验证

• 手动测试：先不带定时执行，直接运行脚本观察输出日志。
• 检查时序数据库：在 InfluxDB 中查询最近写入点：
  

  from(bucket: "cache_metrics") |> range(start: -10m) |> filter(fn: (r) => r._measurement == "cache_hit_rate")

• 模拟异常：将 ALERT_THRESHOLD 设为 0.99，触发告警，观察 Webhook 是否收到 POST。
• 网络连通性：telnet <redis_host> 6379 确认可达。

4.7 常见错误与解决方法

4.8 性能与稳定性优化

• 连接池与复用：避免每次请求新建 TCP 连接。Redis 客户端默认维护连接池；InfluxDB 客户端也内置重连机制。
• 批量写入：如果同时采集多个缓存实例，可以聚合数据点后用 WriteAPI.WritePoints 一次写入，减少 I/O。
• 采集失败重试：网络抖动时，在脚本内加入指数退避重试（最多3次），避免丢失数据点。
• 超时控制：Redis / InfluxDB 连接均设置 3 秒超时，防止脚本堵塞。
• 日志轮转：生产环境将标准输出重定向到文件，配合 logrotate 管理日志。
• 健康检查：在采集脚本中加入 /health HTTP 端点（如果常驻运行），纳入监控系统。