""" Ch13 配套代码 2 / 4 —— INFO 监控大盘 + 健康度评分 演示： 1. 解析 INFO 输出 2. 按 9 大 section 美化打印关键指标 3. 计算健康度评分（满分 100，按 6 大维度扣分） 4. 输出诊断报告 """ import redis r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True) def section(title: str) -> None: print("\n" + "=" * 64) print(title) print("=" * 64) def fmt_bytes(n: int) -> str: if n is None: return "-" n = float(n) for unit in ["B", "KB", "MB", "GB", "TB"]: if n < 1024: return f"{n:.2f}{unit}" n /= 1024 return f"{n:.2f}PB" def kv(label: str, value, unit: str = "", color_warn: bool = False) -> None: mark = "⚠ " if color_warn else " " print(f" {mark}{label:<32} {value}{unit}") def show_server(info: dict) -> None: section("Section 1: SERVER 进程信息") kv("redis_version", info.get("redis_version")) kv("redis_mode", info.get("redis_mode")) kv("os", info.get("os")) kv("process_id", info.get("process_id")) kv("tcp_port", info.get("tcp_port")) kv("uptime_in_days", info.get("uptime_in_days")) def show_clients(info: dict, score: dict) -> None: section("Section 2: CLIENTS 客户端连接") connected = int(info.get("connected_clients", 0)) blocked = int(info.get("blocked_clients", 0)) rejected = int(info.get("rejected_connections", 0) or 0) kv("connected_clients", connected) kv("blocked_clients", blocked, color_warn=blocked > 10) kv("rejected_connections", rejected, color_warn=rejected > 0) kv("maxclients", info.get("maxclients")) if rejected > 0: score["rejected"] = -15 if blocked > 10: score["blocked"] = -5 def show_memory(info: dict, score: dict) -> None: section("Section 3: MEMORY 内存") used = int(info.get("used_memory", 0)) rss = int(info.get("used_memory_rss", 0)) peak = int(info.get("used_memory_peak", 0)) maxmem = int(info.get("maxmemory", 0) or 0) frag = float(info.get("mem_fragmentation_ratio", 1.0)) used_pct = (used / maxmem * 100) if maxmem else 0 kv("used_memory", fmt_bytes(used)) kv("used_memory_rss (OS看)", fmt_bytes(rss)) kv("used_memory_peak", fmt_bytes(peak)) kv("maxmemory", fmt_bytes(maxmem) if maxmem else "无限制（危险！）", color_warn=not maxmem) kv("内存使用率", f"{used_pct:.1f}%", color_warn=used_pct > 80) kv("mem_fragmentation_ratio", f"{frag:.2f}", color_warn=frag > 1.5 or frag < 1.0) if frag > 1.5: score["frag_high"] = -10 if frag < 1.0: score["swap"] = -25 if used_pct > 80: score["mem_high"] = -10 if not maxmem: score["no_maxmem"] = -10 def show_persistence(info: dict, score: dict) -> None: section("Section 4: PERSISTENCE 持久化") fork_us = int(info.get("latest_fork_usec", 0)) aof_on = int(info.get("aof_enabled", 0)) aof_pending = int(info.get("aof_pending_bio_fsync", 0) or 0) aof_delayed = int(info.get("aof_delayed_fsync", 0) or 0) rdb_in_progress = int(info.get("rdb_bgsave_in_progress", 0)) kv("rdb_changes_since_last_save", info.get("rdb_changes_since_last_save")) kv("rdb_bgsave_in_progress", rdb_in_progress) kv("latest_fork_usec", f"{fork_us}us ({fork_us/1000:.1f}ms)", color_warn=fork_us > 500000) kv("aof_enabled", aof_on) if aof_on: kv("aof_current_size", fmt_bytes(int(info.get("aof_current_size", 0)))) kv("aof_pending_bio_fsync", aof_pending, color_warn=aof_pending > 0) kv("aof_delayed_fsync", aof_delayed, color_warn=aof_delayed > 0) if fork_us > 1_000_000: score["fork_slow"] = -15 elif fork_us > 500_000: score["fork_slow"] = -5 if aof_delayed > 0: score["aof_delayed"] = -10 def show_stats(info: dict, score: dict) -> None: section("Section 5: STATS 全局统计") qps = int(info.get("instantaneous_ops_per_sec", 0)) in_kb = float(info.get("instantaneous_input_kbps", 0)) out_kb = float(info.get("instantaneous_output_kbps", 0)) hits = int(info.get("keyspace_hits", 0)) misses = int(info.get("keyspace_misses", 0)) evict = int(info.get("evicted_keys", 0)) expired = int(info.get("expired_keys", 0)) total = hits + misses hit_ratio = (hits / total * 100) if total else 0 kv("instantaneous_ops_per_sec", f"{qps} QPS") kv("instantaneous_input_kbps", f"{in_kb:.2f} KB/s") kv("instantaneous_output_kbps", f"{out_kb:.2f} KB/s") kv("keyspace_hits", hits) kv("keyspace_misses", misses) kv("命中率", f"{hit_ratio:.2f}%", color_warn=hit_ratio < 90 and total > 100) kv("expired_keys", expired) kv("evicted_keys", evict, color_warn=evict > 0) if total > 100 and hit_ratio < 90: score["hit_low"] = -10 if evict > 0: score["evict"] = -5 def show_replication(info: dict) -> None: section("Section 6: REPLICATION 主从复制") role = info.get("role") kv("role", role) if role == "master": kv("connected_slaves", info.get("connected_slaves")) kv("master_repl_offset", info.get("master_repl_offset")) else: kv("master_host", info.get("master_host")) kv("master_link_status", info.get("master_link_status")) kv("master_last_io_seconds_ago", info.get("master_last_io_seconds_ago")) def show_cpu(info: dict) -> None: section("Section 7: CPU") kv("used_cpu_sys", info.get("used_cpu_sys")) kv("used_cpu_user", info.get("used_cpu_user")) kv("used_cpu_sys_children", info.get("used_cpu_sys_children")) kv("used_cpu_user_children", info.get("used_cpu_user_children")) def show_keyspace(info_keyspace: dict) -> None: section("Section 8: KEYSPACE 各 DB 的 Key 数") if not info_keyspace: kv("(empty)", "—") return for db, stats in info_keyspace.items(): kv(db, stats) def health_score(score: dict) -> None: section("Section 9: 健康度评分") base = 100 deduct = sum(score.values()) final = max(0, base + deduct) print(f" 基础分：{base}") if score: for reason, d in score.items(): print(f" {reason:<20} {d:+d}") else: print(" （无扣分项）") print(f"\n 最终得分：{final} / 100") if final >= 90: print(" ✅ 健康") elif final >= 70: print(" ⚠ 亚健康，建议关注") else: print(" ❌ 重病！立即排查") def main() -> None: try: info = r.info() info_ks = r.info("keyspace") score: dict = {} show_server(info) show_clients(info, score) show_memory(info, score) show_persistence(info, score) show_stats(info, score) show_replication(info) show_cpu(info) show_keyspace(info_ks) health_score(score) except redis.ConnectionError as e: print(f"❌ Redis 连接失败：{e}") if __name__ == "__main__": main()