主题
第 13 章 性能优化、监控与排障
学习目标：掌握一套自上而下、有据可依的 Redis 排障方法论，能熟练使用 SLOWLOG / INFO / MEMORY 三大利器定位问题，看懂 Prometheus + Grafana 监控大盘，并能背出 13 条生产环境优化清单。看完之后能从容应对「线上 Redis 突然变慢」「内存涨疯了」「QPS 上不去」这类紧急工单。
13.0 概览：从「玄学排障」到「方法论排障」
很多同学碰到「Redis 慢」第一反应就是「重启大法」「加内存」「换机器」——这是典型的没有方法论。本章建立一套生产可落地的「望闻问切」流程：
┌───────────────────────────────────────────────────────────────────┐
│ Redis 排障 4 层模型（自上而下） │
├───────────────────────────────────────────────────────────────────┤
│ │
│ Layer 1：客户端 ← 连接池？序列化？业务侧批量？ │
│ │ │
│ ▼ │
│ Layer 2：网络 ← RTT？带宽？TCP 重传？ │
│ │ │
│ ▼ │
│ Layer 3：Redis 主线程 ← 慢命令？大 Key？fork？AOF fsync？ │
│ │ │
│ ▼ │
│ Layer 4：操作系统 ← CPU 满？SWAP？THP？磁盘 IO？ │
│ │
└───────────────────────────────────────────────────────────────────┘
📌 排障原则：先定位 Layer，再下钻到具体指标🩺 生活类比：医生看病不会一上来就开刀，而是先「望」（看气色 = 业务监控）→「闻」（听呼吸 = 慢日志）→「问」（问症状 = INFO）→「切」（号脉 = 火焰图）。Redis 排障也一样，先收集证据再下结论。
本章配套实战 4 个脚本和 1 个交互演示页面，建议边看边跑。
13.1 排障的整体方法论：自上而下 4 层定位
13.1.1 Layer 1：客户端——80% 的问题其实在自己代码里
很多「Redis 慢」其实是客户端用法错误：
┌────────────────────────┬─────────────────────────────────────┐
│ 现象 │ 根因 │
├────────────────────────┼─────────────────────────────────────┤
│ 一个简单 GET 耗时 50ms │ 没用连接池，每次都 TCP 三次握手 │
│ 批量插入 1w 条要 30s │ 没用 Pipeline，1w 次 RTT │
│ JSON 序列化耗时 │ 客户端 CPU 瓶颈，不是 Redis 慢 │
│ 连接数报警 │ 用完没还给池，连接泄漏 │
└────────────────────────┴─────────────────────────────────────┘第一步永远是问自己 4 个问题：
- 用连接池了吗？
- 用 Pipeline / MGET 了吗？
- 客户端 CPU 是不是打满了？（序列化、解压都是 CPU）
- 业务代码里是不是有循环里的
for k in keys: r.get(k)？
13.1.2 Layer 2：网络——RTT 是物理上限
Redis 单次往返时延 = 网络 RTT + Redis 处理时间。同机房 0.1~0.5ms，跨机房可达 5~30ms。Pipeline 的本质就是省 RTT。
排障命令：
bash
# 测延迟
redis-cli --latency -h 127.0.0.1
redis-cli --latency-history -h 127.0.0.1 # 历史模式
redis-cli --latency-dist -h 127.0.0.1 # 分布图
redis-cli --intrinsic-latency 5 # 测系统本身的内核调度延迟
# 测网络
ping <redis-host>
mtr <redis-host> # 排查中间路由跳点
ss -ti # 看 TCP 重传/RTT💡 一个常被忽略的细节：同 VPC 不同可用区之间也有 ms 级延迟，对延迟敏感的业务要把客户端和 Redis 部署到同 AZ。
13.1.3 Layer 3：Redis 主线程——本章重点
Redis 是单线程处理命令的（Redis 6 之后 IO 多线程，但命令执行依然单线程）。意味着：任何一条慢命令都会阻塞所有其他请求。
Redis 主线程时间轴
┌────┬──────────────────┬────┬────┐
│GET │ KEYS * (200ms) │SET │GET │
└────┴──────────────────┴────┴────┘
↑
这 200ms 内所有客户端排队等排查 Layer 3 的核心 4 件套：
- SLOWLOG ：慢命令记录器（13.2）
- INFO：全维度指标快照（13.6）
- MONITOR：命令录像机（开发可用，生产严禁）
- LATENCY：延迟事件历史
13.1.4 Layer 4：操作系统——基础设施异常
Redis 也是 Linux 进程，OS 的问题直接传导：
┌──────────────────┬──────────────────────────────────────┐
│ OS 异常 │ 对 Redis 的影响 │
├──────────────────┼──────────────────────────────────────┤
│ SWAP 触发 │ 任意命令延迟飙升 100x │
│ THP（透明大页） │ fork 时 COW 复制变慢 │
│ vm.overcommit_memory=0 │ fork 时报 OOM 直接挂 │
│ CPU 软中断飙高 │ 网络包处理跟不上 │
│ 磁盘 await 高 │ AOF fsync 阻塞主线程 │
└──────────────────┴──────────────────────────────────────┘排查命令：top、vmstat 1、iostat -x 1、pidstat -d 1、cat /sys/kernel/mm/transparent_hugepage/enabled。
13.2 慢查询日志 SLOWLOG
Redis 自带的「慢命令记录器」，类似 MySQL 的 slow_log。线上必开。
13.2.1 配置
bash
# 配置文件 redis.conf
slowlog-log-slower-than 10000 # 单位微秒，默认 10000us = 10ms
slowlog-max-len 128 # 最多保留 128 条（环形缓冲）
# 在线修改（不用重启）
CONFIG SET slowlog-log-slower-than 1000 # 改成 1ms
CONFIG SET slowlog-max-len 1024
CONFIG REWRITE # 持久化到配置文件💡 生产建议：阈值设为 1ms (1000us)——10ms 太宽松，绝大多数慢命令会漏掉；保留条数设为 1024，便于回溯。
13.2.2 命令
bash
SLOWLOG GET # 取全部
SLOWLOG GET 10 # 取最近 10 条
SLOWLOG LEN # 记录条数
SLOWLOG RESET # 清空
SLOWLOG HELP # 帮助13.2.3 字段解读
127.0.0.1:6379> SLOWLOG GET 1
1) 1) (integer) 14 ← id（自增）
2) (integer) 1713244800 ← UNIX 时间戳
3) (integer) 213000 ← 执行耗时（微秒）= 213ms
4) 1) "KEYS" ← 命令
2) "*"
5) "127.0.0.1:51234" ← 客户端 ip:port（4.0+）
6) "service-payment" ← 客户端 name（CLIENT SETNAME）6 个字段记忆口诀：「ID-时间-耗时-命令-哪台机-哪个服务」。重点关注第 5、6 字段——能直接定位到肇事业务方。
13.2.4 实战流程
1. 报警触发 → SSH 到机器
2. redis-cli SLOWLOG GET 50 → 找耗时 Top 的命令
3. 看命令参数 → 定位「大 Key」「KEYS *」「LRANGE 大 List」等
4. 看客户端 IP → 找出业务方
5. 通知业务方修改（用 SCAN/HSCAN/分批/Pipeline 替代）
6. SLOWLOG RESET → 验证修改后是否还有13.3 大 Key / 热 Key 排查
13.3.1 大 Key 的危害（回顾）
第 4 章讲过，这里只回顾结论：
- 网络阻塞：单 Key MB 级别，传输打满网卡
- 主线程卡顿：DEL 一个 1GB 的 Hash 能阻塞 1s+
- 集群迁移失败：CLUSTER 槽位迁移单 Key 不可拆分
- 持久化抖动：RDB / AOF 写入大 Key 时慢
13.3.2 大 Key 排查工具链
① redis-cli --bigkeys
bash
redis-cli -h <host> --bigkeys
# 实时扫描，对每种类型输出最大的 Key输出示例：
[00.00%] Biggest hash found so far 'user:profile:1001' with 50000 fields
[03.21%] Biggest list found so far 'queue:order' with 120000 items
...
-------- summary -------
Biggest string found 'session:xx' has 8294400 bytes
Biggest list found 'queue:order' has 120000 items
Biggest hash found 'user:profile:1001' has 50000 fields⚠️
--bigkeys是采样性质的（用 SCAN 遍历），结果是「最大的」之一，不是绝对最大。 ⚠️ 它对每种类型的「大」定义不同：String 看字节数，其他看元素数，不能直接比内存。
② redis-cli --memkeys（5.0+）
bash
redis-cli --memkeys # 按内存占用算
redis-cli --memkeys-samples 0 # 全量扫描比 --bigkeys 更准——直接调 MEMORY USAGE，所有类型统一按字节数比较。
③ rdb-tools 离线分析
bash
pip install rdbtools
rdb -c memory dump.rdb --bytes 10240 -f bigkey.csv
# 找出占用 > 10KB 的 Key，输出 CSV强烈推荐生产用法：每天夜里下载 RDB 离线分析，完全零开销。
④ 业务侧拦截（最优解）
在客户端 SDK 里加拦截器：
python
def safe_set(r, key, value, max_size=10240):
if len(value) > max_size:
raise ValueError(f"big key denied: {key} size={len(value)}")
r.set(key, value)「事后排查」不如「事前拦截」——架构组应该在公司基础库里默认禁止 1MB 以上的写入。
13.3.3 热 Key 排查
热 Key（Hot Key）：访问频次极高的 Key，会让单分片 CPU 打满。
| 工具 | 优缺点 |
|---|---|
redis-cli --hotkeys | 需要 maxmemory-policy=allkeys-lfu；轻量但需配置 |
MONITOR | 简单粗暴；生产严禁（QPS 减半） |
| Facebook redis-faina | 解析 MONITOR 输出做统计，开发利器 |
| 客户端打点 | 在 SDK 里上报每次 GET 的 key 哈希；最稳健 |
| 抓包分析 | tcpdump + 自研解析；离线 |
| Proxy 层统计 | 有 Codis / Twemproxy 的可以直接出报表 |
redis-cli --hotkeys 实战：
bash
# 先改策略
redis-cli CONFIG SET maxmemory-policy allkeys-lfu
# 等几分钟让 LFU 计数器累积
redis-cli --hotkeys13.3.4 治理方案
┌───────────────────┬──────────────────────────────────────┐
│ 方案 │ 适用场景 │
├───────────────────┼──────────────────────────────────────┤
│ 拆分大 Key │ user:1 → user:1:base / user:1:settings │
│ 本地缓存（多级缓存） │ 读多写少的热 Key（首页配置） │
│ 读写分离 │ 主写从读；适合写少 │
│ 多副本散列 │ key_1 / key_2 / ... / key_N，随机访问 │
│ 客户端分片 │ 大 Hash 哈希取模到 N 个 Hash │
│ 改用 ZSet/SortedSet │ 需要排行榜场景拆分 │
└───────────────────┴──────────────────────────────────────┘13.4 BIO 后台线程
13.4.1 为什么需要后台线程
Redis 4.0 之前是「纯」单线程，所有 IO 都同步做。结果：
- AOF fsync 调用
fsync(fd)时阻塞主线程 - DEL 大 Key 时释放内存阻塞主线程
- 关闭被动连接的 fd 时也可能阻塞
Redis 4.0 引入 BIO（Background IO）线程，把这些「可异步」的工作扔到后台。
┌─────────────────────────────────────────────────────────┐
│ Redis 4.0+ 线程模型 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ │
│ │ 主线程 │ ← 命令执行、IO 多路复用 │
│ └──────┬───────┘ │
│ │ │
│ │ 投递任务 │
│ ▼ │
│ ┌──────────────────────────────────────────┐ │
│ │ BIO 任务队列（按类型分 3 个队列） │ │
│ ├──────────────────────────────────────────┤ │
│ │ ① BIO_CLOSE_FILE ← 关闭被动 fd │ │
│ │ ② BIO_AOF_FSYNC ← AOF fsync │ │
│ │ ③ BIO_LAZY_FREE ← UNLINK / 异步释放 │ │
│ └────┬──────────┬──────────┬───────────────┘ │
│ ▼ ▼ ▼ │
│ [bio thread] [bio thread] [bio thread] │
│ │
└─────────────────────────────────────────────────────────┘13.4.2 异步删除：UNLINK 和 lazy free
bash
DEL bigkey # 同步删，可能阻塞主线程几百 ms
UNLINK bigkey # 异步删！主线程只解链，真正释放在 BIO 线程老 Redis 想异步删 Hash/Set 元素，可以打开「lazy free」：
bash
CONFIG SET lazyfree-lazy-eviction yes # 内存淘汰用异步
CONFIG SET lazyfree-lazy-expire yes # 过期清理用异步
CONFIG SET lazyfree-lazy-server-del yes # 隐式删除（如 RENAME 覆盖）
CONFIG SET lazyfree-lazy-user-del yes # 让 DEL 也走异步13.4.3 怎么观察后台线程是否积压
bash
INFO persistence | grep -E "aof_(pending|delayed_fsync)"
# aof_pending_bio_fsync ← 待 fsync 的请求数（积压则警告）
# aof_delayed_fsync ← 因 fsync 慢被延后的次数
INFO stats | grep lazyfree
# lazyfree_pending_objects ← 待异步释放的对象数
INFO persistence | grep latest_fork_usec
# latest_fork_usec ← 上次 fork 耗时（微秒）⚠️ 如果
aof_delayed_fsync持续增长，说明磁盘 IO 跟不上——该换 SSD 了。
13.5 阻塞主线程的「定时炸弹」
13.5.1 慢命令清单（背下来）
┌──────────────────────┬─────────┬───────────────────────────┐
│ 危险命令 │ 复杂度 │ 替代方案 │
├──────────────────────┼─────────┼───────────────────────────┤
│ KEYS pattern │ O(N) │ SCAN（游标式遍历） │
│ HGETALL big_hash │ O(N) │ HSCAN + 限制单次返回数 │
│ SMEMBERS big_set │ O(N) │ SSCAN │
│ SORT big_list │ O(N+M log M) │ 业务侧排序 │
│ SUNIONSTORE big_set │ O(N) │ 拆批处理 │
│ ZRANGE 0 -1 │ O(N+M) │ 分页 ZRANGE 0 99 │
│ FLUSHALL / FLUSHDB │ O(N) │ FLUSHDB ASYNC │
│ LRANGE 0 -1 big_list │ O(N+M) │ 分页 │
│ EVAL 长 Lua 脚本 │ 取决于脚本 │ 拆分逻辑 / 设置超时 │
│ DEBUG SLEEP n │ O(N) │ 仅用于测试 │
└──────────────────────┴─────────┴───────────────────────────┘13.5.2 用 DEBUG SLEEP 测试主线程被阻塞的影响
开两个终端：
bash
# 终端 A：模拟慢命令
redis-cli DEBUG SLEEP 5
# 终端 B：尝试任意操作
redis-cli SET foo bar
# 会卡 5 秒才返回 OK！🧨 这就是单线程的代价——一个慢命令把整个 Redis 「冻住」。
13.5.3 SCAN 系列：渐进式遍历的正确姿势
bash
SCAN 0 MATCH user:* COUNT 100
# 返回：[next_cursor, [k1, k2, ...]]
# 持续调用直到 next_cursor = 0SCAN 的特性：
- 游标式，不会一次返回所有 Key
- 单次 O(1)；总开销 O(N)，但分散到多次调用
- 可能返回重复元素，业务侧需自行去重
- COUNT 是「建议值」，不是精确值
Python 示例：
python
for k in r.scan_iter(match="user:*", count=200):
process(k) # 一次只阻塞 200 个 key 的扫描时间SCAN 家族：SCAN（顶层 Key）/ HSCAN（Hash）/ SSCAN（Set）/ ZSCAN（ZSet）。
13.6 INFO 命令详解（重点）
INFO 是 Redis 的「全身体检报告」，所有监控系统底层都靠它。
13.6.1 9 大 section
bash
INFO server # 进程基本信息
INFO clients # 客户端连接
INFO memory # 内存
INFO persistence # RDB / AOF
INFO stats # 全局统计
INFO replication # 主从复制
INFO cpu # CPU 占用
INFO commandstats # 每个命令的累计耗时
INFO cluster # 集群信息
INFO keyspace # 每个 db 的 key 数
INFO # 全部
INFO all # 全部 + 包含默认隐藏的 commandstats / latencystats13.6.2 关键监控指标（必背）
内存类
used_memory ← Redis 申请的内存（包含碎片）
used_memory_rss ← OS 视角看 Redis 实际占用（RSS）
used_memory_peak ← 历史峰值
mem_fragmentation_ratio ← 碎片率 = rss / used_memory
maxmemory ← 上限碎片率解读：
> 1.5 严重碎片，考虑重启或 MEMORY PURGE
1.0~1.5 正常
< 1.0 部分内存被换到 SWAP（最危险！）💡 Redis 4.0+ 支持主动碎片整理：
CONFIG SET activedefrag yes。
吞吐与连接
instantaneous_ops_per_sec ← 当前 QPS（实时）
instantaneous_input_kbps ← 当前入向带宽（KB/s）
instantaneous_output_kbps ← 当前出向带宽
total_connections_received ← 累计连接数
connected_clients ← 当前在线客户端
rejected_connections ← 因超 maxclients 被拒绝的连接数（非 0 报警）
blocked_clients ← 阻塞客户端数（BLPOP / WAIT 等）命中率
keyspace_hits ← 命中次数
keyspace_misses ← 未命中次数
命中率 = hits / (hits + misses) ← 应 ≥ 95%python
hit_ratio = hits / (hits + misses) if (hits + misses) else 0过期 / 淘汰
expired_keys ← 累计被过期清理的 key 数
evicted_keys ← 累计被淘汰的 key 数（非 0 说明 maxmemory 不够）Fork 相关（最容易被忽略）
latest_fork_usec ← 上次 fork 耗时（微秒），> 1_000_000 说明 1s+
rdb_last_save_time
rdb_changes_since_last_save
aof_current_size
aof_base_size⚠️ fork > 500ms 就要报警，> 1s 必须查根因（可能是大内存 + THP 没关）。
CPU
used_cpu_sys ← 内核态 CPU 时间
used_cpu_user ← 用户态 CPU 时间将相邻两次采样做差，可以算出 CPU 占用百分比。
13.6.3 INFO 的查看小技巧
bash
# 只看一行
redis-cli INFO memory | grep used_memory_human
# 用 redis-cli --stat 看一个滚动统计
redis-cli --stat
# ------- data ------ --------------------- load ------------ - child -
# keys mem clients blocked requests connections
# 18 1.05M 1 0 157 (+0) 313.7 监控系统接入
13.7.1 Prometheus + redis_exporter
bash
# 启动 redis_exporter（开源项目 oliver006/redis_exporter）
docker run -d -p 9121:9121 \
-e REDIS_ADDR=redis://your-host:6379 \
-e REDIS_PASSWORD=xxx \
oliver006/redis_exporter
# Prometheus scrape config
scrape_configs:
- job_name: redis
static_configs:
- targets: ['redis-exporter:9121']redis_exporter 会自动把 INFO 的所有指标转成 Prometheus metrics（命名以 redis_ 开头）。
13.7.2 Grafana Dashboard
开源大盘 ID（直接在 Grafana 导入）：
- 17507 — 综合 Redis Dashboard（推荐入门）
- 763 — 老牌 Redis Dashboard
- 11835 — Redis Cluster Dashboard
- 15740 — Redis Sentinel Dashboard
13.7.3 告警规则示例（Prometheus rule）
yaml
groups:
- name: redis
rules:
- alert: RedisMemoryHigh
expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.8
for: 5m
annotations: { summary: "Redis 内存使用率 > 80%" }
- alert: RedisHitRatioLow
expr: rate(redis_keyspace_hits_total[5m]) /
(rate(redis_keyspace_hits_total[5m]) + rate(redis_keyspace_misses_total[5m])) < 0.9
for: 10m
annotations: { summary: "Redis 命中率 < 90%" }
- alert: RedisForkSlow
expr: redis_latest_fork_usec > 500000
for: 1m
annotations: { summary: "Redis fork 耗时 > 500ms" }
- alert: RedisRejectedConn
expr: increase(redis_rejected_connections_total[5m]) > 0
annotations: { summary: "有连接被拒绝（maxclients 不够）" }
- alert: RedisEvictionRising
expr: rate(redis_evicted_keys_total[5m]) > 100
annotations: { summary: "Redis 淘汰速率过高（内存不够）" }🎯 告警分级：内存 / fork / 命中率 → P1；连接拒绝 / 淘汰 → P2；QPS 突变 → P3。
13.8 性能优化清单（13 条）
这 13 条是「生产经过验证」的最佳实践，全部命中能让 P99 延迟下降一个量级。
┌───┬─────────────────────────────┬─────────────────────────┐
│ # │ 优化项 │ 预期收益 │
├───┼─────────────────────────────┼─────────────────────────┤
│ 1 │ 客户端用连接池 │ 单次 GET 从 5ms → 0.5ms │
│ 2 │ 用 Pipeline / MGET / HMGET │ 批量场景 10x ~ 100x │
│ 3 │ Lua 下沉「读-改-写」复合逻辑 │ 减少 RTT、保证原子 │
│ 4 │ 设置合理 TTL，避免内存堆积 │ 内存稳定 / 命中率提升 │
│ 5 │ 用 listpack 友好的小数据结构 │ 内存节省 50%+，速度更快 │
│ 6 │ 控制 Key 长度 │ 节省内存 + 哈希更均匀 │
│ 7 │ 禁用 KEYS / FLUSHALL │ 杜绝主线程长时间阻塞 │
│ 8 │ 用 UNLINK 替代 DEL │ 大 Key 删除不阻塞 │
│ 9 │ 拆分大 Key │ 网络 / 集群迁移友好 │
│10 │ 业务隔离（不同业务不同实例） │ 避免互相影响 │
│11 │ 错峰主从全量同步 │ fork 不冲击业务高峰 │
│12 │ maxmemory 留 30% 给 fork │ 不 OOM、不 SWAP │
│13 │ Linux 调优 │ overcommit / THP / SWAP │
└───┴─────────────────────────────┴─────────────────────────┘1. 客户端用连接池
python
import redis
pool = redis.ConnectionPool(host="...", max_connections=100)
r = redis.Redis(connection_pool=pool)2. Pipeline / MGET / HMGET
python
pipe = r.pipeline(transaction=False)
for k in keys:
pipe.get(k)
results = pipe.execute() # 一次 RTT 拿全部3. Lua 下沉
lua
-- check_and_decr.lua
local stock = tonumber(redis.call('GET', KEYS[1]))
if stock and stock > 0 then
redis.call('DECR', KEYS[1])
return 1
end
return 04. TTL：永远要给缓存设过期时间
bash
SET cache:user:1 "..." EX 3600
HSET ... + EXPIRE # Hash 的 TTL 是整个 Key 级别的5. listpack 友好
让 Hash / Set / ZSet 的元素数 < 128 且元素长度 < 64 字节，能保持紧凑编码，内存节省 50% 以上。
6. 控制 Key 长度
不要写 user:profile:detail:settings:notification:email:enabled:1001， 推荐 up:nfn:e:1001 之类短代号。百万 Key 累积下来差几个 GB。
7. 禁用危险命令
bash
# redis.conf 重命名为空字符串 = 彻底禁用
rename-command KEYS ""
rename-command FLUSHALL ""
rename-command FLUSHDB ""
rename-command CONFIG "CONFIG_a8b3f2" # 改名而非禁用
rename-command DEBUG ""8. UNLINK 替代 DEL
业务代码里所有 DEL 都改成 UNLINK（除非业务依赖 DEL 的同步语义）。
9. 拆分大 Key
大 Hash 哈希分片：
python
def field_to_shard(field, n=16):
return crc32(field) % n
key = f"user:profile:1001:{field_to_shard(field)}"10. 业务隔离
坚决不要把业务 A 和业务 B 塞同一个 Redis 实例。一个业务出大 Key 把整个实例阻塞，全公司业务连坐。
11. 错峰主从同步
bash
CONFIG SET save "" # 关闭 RDB 自动触发
# 凌晨 3 点 cron 触发 BGSAVE12. maxmemory 留 30% 给 fork
机器 32GB → maxmemory 设 22GB（不是 30GB）。fork 时 COW 最坏会再用一份内存。
13. Linux 调优清单
bash
# 必做
echo 1 > /proc/sys/vm/overcommit_memory # fork 不报 OOM
echo never > /sys/kernel/mm/transparent_hugepage/enabled # 关闭 THP
sysctl -w net.core.somaxconn=65535 # TCP backlog
sysctl -w vm.swappiness=1 # 不到迫不得已不 swap
# 持久化到 /etc/sysctl.conf
echo "vm.overcommit_memory=1" >> /etc/sysctl.conf
echo "net.core.somaxconn=65535" >> /etc/sysctl.conf
echo "vm.swappiness=1" >> /etc/sysctl.conf
sysctl -p13.9 实操：跑一遍配套代码
实战代码见 13_optimization/code/：
01_slowlog_analyzer.py：触发 DEBUG SLEEP / LRANGE 大 List，再 SLOWLOG GET 解析展示02_info_dashboard.py：解析 INFO 输出，按 section 美化打印 + 健康度评分03_memory_analyzer.py：MEMORY STATS / USAGE / OBJECT IDLETIME 综合分析04_optimization_checklist.py：自动检查 Redis 配置 vs 13 条最佳实践
浏览器演示见 13_optimization/demo.html：
- ① INFO 监控大盘：模拟实例 + 进度条/仪表盘 + 一键模拟内存暴涨/fork 慢
- ② 慢查询排查实验室：点命令按钮看模拟耗时 + SLOWLOG 记录 + 优化建议
- ③ 13 条优化清单 self-check：勾选当前配置自动打分 + Top 3 优化建议
13.10 本章小结
┌───────────────────────────────────────────────────────────┐
│ 本章核心要点 │
├───────────────────────────────────────────────────────────┤
│ │
│ ① 排障 4 层模型：客户端 → 网络 → 主线程 → 操作系统 │
│ │
│ ② SLOWLOG 必开，阈值设 1ms，6 个字段重点看 IP & 命令 │
│ │
│ ③ 大 Key 排查：--bigkeys / --memkeys / rdbtools / 业务拦截 │
│ │
│ ④ 热 Key 排查：--hotkeys / 客户端打点 / 多副本散列治理 │
│ │
│ ⑤ BIO 后台线程：异步 fsync / lazy free / UNLINK │
│ │
│ ⑥ 慢命令清单：KEYS / HGETALL / SMEMBERS 用 SCAN 替代 │
│ │
│ ⑦ INFO 6 个核心指标：碎片率 / QPS / 命中率 / 拒绝连接 / │
│ fork 耗时 / 淘汰速率 │
│ │
│ ⑧ 监控选 Prometheus + redis_exporter + Grafana 17507 │
│ │
│ ⑨ 13 条优化清单：连接池 / Pipeline / Lua / TTL / listpack / │
│ 短 Key / 禁危险命令 / UNLINK / 拆 Key / 隔离 / │
│ 错峰同步 / 留 30% fork / Linux 调优 │
│ │
└───────────────────────────────────────────────────────────┘13.11 面试高频题
Q1：怎么定位 Redis 变慢？给出排查思路
考察点：是否有方法论，而不是「重启大法」。
标准答案——按 4 层模型自上而下：
- 客户端层：先看业务监控 P99 延迟、RT 分布；看是不是只有某个业务慢；查代码有没有 N+1（循环里调 GET）、有没有用连接池、有没有 Pipeline。
- 网络层：
redis-cli --latency测延迟；ping/mtr看 RTT；ss -ti看 TCP 重传率。 - Redis 主线程层：
SLOWLOG GET 50找慢命令INFO commandstats看哪个命令累计耗时高INFO memory看碎片率、used_memory_rss是否异常INFO persistence看latest_fork_usec、aof_delayed_fsyncCLIENT LIST看是否有阻塞客户端redis-cli --bigkeys/--hotkeys排查大热 Key
- OS 层：
top看 CPU；vmstat 1看 SWAP；iostat -x 1看磁盘 await；查 THP 是否关闭。
加分项：提到 LATENCY DOCTOR 命令（Redis 内置的「医生」会给出排查建议）。
Q2：SLOWLOG 怎么用？看哪些字段？
考察点：实战经验。
标准答案：
配置：
slowlog-log-slower-than：阈值（微秒），生产建议 1000us = 1msslowlog-max-len：环形缓冲条数，建议 1024- 可在线
CONFIG SET修改
命令：
SLOWLOG GET [N]：取最近 N 条SLOWLOG LEN：当前条数SLOWLOG RESET：清空
字段（共 6 个）：
- id（自增）
- UNIX 时间戳
- 执行耗时（微秒）—— 这个数字越大越严重
- 命令及参数 —— 看是不是 KEYS / HGETALL / LRANGE 等
- 客户端 ip:port —— 直接定位肇事方
- 客户端 name（CLIENT SETNAME 设置的）—— 配合 CMDB 找业务
加分项：提到 SLOWLOG 是纯内存的环形缓冲，不会持久化，重启就丢；要长期保留需要监控系统定期采集。
Q3：内存碎片率多少算正常？怎么处理？
考察点：内存底层理解。
标准答案：
指标定义：mem_fragmentation_ratio = used_memory_rss / used_memory
判断：
< 1.0 SWAP 触发！RSS < 内存 = 部分被换到磁盘，必须立即处理
1.0~1.5 正常范围
> 1.5 严重碎片，需要处理碎片产生原因：jemalloc 按固定 size class 分配（8B、16B、32B、64B...），频繁 SET/DEL 不同大小的 Key 会留下空洞。
处理方式：
- 主动碎片整理（推荐）：Redis 4.0+
CONFIG SET activedefrag yes CONFIG SET active-defrag-ignore-bytes 100mb CONFIG SET active-defrag-threshold-lower 10 CONFIG SET active-defrag-threshold-upper 100 - 手动：
MEMORY PURGE（仅释放 jemalloc 内部碎片，不一定全释放）。 - 重启大法：从节点先重启，再切换主从。
加分项：解释 jemalloc 的 size class 分级、为什么 SWAP 触发会让 Redis 慢 100x（内存→磁盘性能差距）。
Q4：Redis 为什么要 fork？fork 耗时长的原因？
考察点：持久化 + Linux 内存模型。
标准答案：
为什么要 fork：RDB / AOF 重写需要一份「数据快照」。直接遍历主进程数据会阻塞主线程；用 fork() 创建子进程，利用 COW（Copy On Write）让子进程「秒级」拥有快照，主进程继续服务。
fork 耗时的原因：
- fork 不是真的复制内存，但要复制页表。
- 页表大小约为
内存大小 / 4096 * 8 字节（每页 4KB，每个 PTE 8 字节）。 - 32GB Redis 实例的页表约 64MB，复制页表耗时约 几百 ms 到 1s+。
优化：
- 关闭 THP（透明大页）：THP 让 PTE 变成 2MB 一个，节省 PTE 数量；但 THP 触发 COW 时是 2MB 一次复制（vs 4KB 一次），写少量数据会放大 IO 500x。所以必须关闭。bash
echo never > /sys/kernel/mm/transparent_hugepage/enabled - 不要把实例搞太大：单实例建议 < 10GB。
vm.overcommit_memory=1：fork 时 Linux 默认会检查「物理内存够不够再复制一遍」，不够直接 OOM 让 fork 失败。开 overcommit 之后允许 fork 成功，COW 时再实际分配。- 错峰：避开业务高峰再 BGSAVE。
- 监控
latest_fork_usec：> 500ms 就报警。
加分项：提到「fork 不阻塞主线程的网络 IO，但调用 fork() 本身那一瞬间是阻塞的」——这就是 fork 慢会导致延迟尖刺的原因。
Q5：哪些命令会阻塞主线程？怎么避免？
考察点：单线程模型 + 慢命令清单。
标准答案：
典型阻塞命令：
| 命令 | 复杂度 | 替代 |
|---|---|---|
| KEYS pattern | O(N) | SCAN |
| HGETALL big | O(N) | HSCAN + 分批 |
| SMEMBERS big | O(N) | SSCAN |
| LRANGE 0 -1 | O(N+M) | 分页 LRANGE |
| SORT big | O(N+M log M) | 业务侧排序 |
| FLUSHALL/FLUSHDB | O(N) | FLUSHDB ASYNC |
| DEL big_key | O(N) | UNLINK |
| EVAL 长 Lua | 视脚本 | 拆分 / 设超时 |
| DEBUG SLEEP n | O(N) | 测试用 |
| 大集合的 SUNIONSTORE/SINTERSTORE | O(N) | 拆分 |
避免方式：
- SDK 拦截层禁用危险命令
- 大 Key 拆分 + 本地缓存
- 用 SCAN 系列代替遍历命令
- UNLINK 代替 DEL
- Lua 脚本要带超时（
script-timeout） - 重要命令在服务端
rename-command重命名为空字符串
加分项：提 Redis 6 的 IO 多线程只解决了网络 IO 多线程，命令执行依然单线程；并提到 latency-monitor-threshold 配置，可以让 Redis 自己记录延迟事件（LATENCY HISTORY / LATENCY DOCTOR）。
Q6：一个 8 核 32G 的 Redis 单机最多能扛多少 QPS？
考察点：对 Redis 性能有量级感。
标准答案：
理论上限：
- 简单 GET/SET：单实例 8~10w QPS（Redis 单线程主要受 CPU 单核限制）
- 用 IO 多线程（Redis 6.0+，
io-threads 4）可达 15~20w QPS - 网络满千兆（125MB/s），按平均 100B 一次响应：125w QPS 是网卡上限
- 实测：Redis 官方 benchmark 在主流硬件上 GET 可到 10w QPS / 单线程
32GB 内存可存数据量：
- 单 Key + Value 平均 100B、含 redisObject 等开销约 150B
- 可存约 2 亿条简单 KV
- 但真实业务里 Hash / List 等结构开销更大，按 1KB/条算，约 3000w 条
实际生产经验：
- 稳定 QPS 不要超过理论上限的 60%：8~10w 留余量到 5w
- 8 核机器：1 核给主线程 + 4 核给 IO 线程 + 剩余给系统/AOF/网络
- 维度估算：8 核 32G 机器 ≈ 5~10w QPS / 30~50GB 内存利用率上限
加分项：
- 提到「QPS 不止看绝对数字，还要看命令复杂度」：MGET 100 个 key 算 1 个 QPS 还是 100 个？
- 提到「单实例瓶颈在单核」，所以集群水平扩展是高 QPS 的唯一答案
- 真实场景下要结合
INFO commandstats看实际「命令耗时累计」反推可承载 QPS
📌 下一章预告：第 14 章我们用前面 13 章的所有知识，手撸一个完整的实战项目——可能是秒杀系统、短链服务或 Feed 流，把 String/ZSet/Lua/Pipeline/缓存设计/集群/监控全部串起来，让你毕业即「Redis 中级工程师」。
🎬 可视化演示
演示加载缓慢或样式异常？点此在新标签页打开 ↗
💻 示例代码
python
"""
Ch13 配套代码 1 / 4 —— 慢查询日志分析器
演示：
1. 配置 SLOWLOG 阈值
2. 故意触发几个慢命令（DEBUG SLEEP、LRANGE 大 List）
3. SLOWLOG GET 解析 6 个字段，按耗时排序输出 Top N
4. 给出每条慢命令的优化建议
"""
import time
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 setup_slowlog(threshold_us: int = 1000, max_len: int = 1024) -> None:
section(f"Step 1: 配置 SLOWLOG 阈值 = {threshold_us}us，容量 = {max_len}")
r.config_set("slowlog-log-slower-than", threshold_us)
r.config_set("slowlog-max-len", max_len)
r.execute_command("SLOWLOG", "RESET")
print(f" 当前 SLOWLOG 长度：{r.execute_command('SLOWLOG', 'LEN')}")
def trigger_slow_commands() -> None:
section("Step 2: 故意触发慢命令（请稍等 ~6 秒）")
print(" - 模拟 DEBUG SLEEP 0.05（50ms）")
try:
r.execute_command("DEBUG", "SLEEP", "0.05")
except redis.ResponseError as e:
print(f" DEBUG 命令被禁用：{e}")
print(" - 准备一个 5w 元素的 List，再 LRANGE 0 -1")
big_list = "demo:slowlog:biglist"
r.delete(big_list)
pipe = r.pipeline(transaction=False)
for i in range(50000):
pipe.rpush(big_list, f"item-{i}")
pipe.execute()
r.lrange(big_list, 0, -1)
print(" - 准备一个 1w 字段的 Hash，再 HGETALL")
big_hash = "demo:slowlog:bighash"
r.delete(big_hash)
mapping = {f"f{i}": f"v{i}" for i in range(10000)}
r.hset(big_hash, mapping=mapping)
r.hgetall(big_hash)
print(" - 模拟 KEYS *（不要在生产用！）")
r.keys("*")
r.delete(big_list, big_hash)
SUGGESTION_RULES = [
("KEYS", "✗ 全库扫描，O(N)。改用 SCAN 0 MATCH pattern COUNT 100"),
("HGETALL", "✗ 大 Hash 全量返回。改用 HSCAN 渐进式遍历，或拆分 Hash"),
("SMEMBERS", "✗ 大 Set 全量返回。改用 SSCAN 渐进式遍历"),
("LRANGE", "✗ 大 List 范围读。LRANGE 0 -1 改成分页 LRANGE 0 99"),
("SORT", "✗ 服务端排序复杂度高。建议业务侧排序"),
("FLUSHALL", "✗ 清空全库。生产应禁用，或改为 FLUSHDB ASYNC"),
("FLUSHDB", "✗ 清空当前 db。改为 FLUSHDB ASYNC（4.0+）"),
("DEL", "△ 大 Key DEL 阻塞主线程。改用 UNLINK"),
("DEBUG", "△ DEBUG 命令仅开发环境，生产应 rename-command 禁用"),
("SUNIONSTORE", "△ 大集合求并集 O(N)。考虑拆批"),
("SINTERSTORE", "△ 大集合求交集 O(N)。考虑拆批"),
("EVAL", "△ Lua 脚本未限时，可能长时间阻塞主线程"),
]
def suggest(cmd: str) -> str:
upper = cmd.upper()
for keyword, msg in SUGGESTION_RULES:
if keyword in upper:
return msg
return "（无特定建议，关注耗时是否合理）"
def show_slowlog(top_n: int = 10) -> None:
section(f"Step 3: SLOWLOG GET —— 取耗时 Top {top_n}")
raw = r.execute_command("SLOWLOG", "GET", top_n)
if not raw:
print(" （SLOWLOG 为空，可能是阈值太高或没有触发）")
return
rows = []
for entry in raw:
sid = entry[0]
ts = entry[1]
duration = entry[2]
cmd_parts = [str(p) for p in entry[3]]
client_ip = entry[4] if len(entry) > 4 else "-"
client_nm = entry[5] if len(entry) > 5 else "-"
cmd_str = " ".join(cmd_parts)
if len(cmd_str) > 60:
cmd_str = cmd_str[:57] + "..."
rows.append((sid, ts, duration, cmd_str, client_ip, client_nm))
rows.sort(key=lambda x: -x[2])
print(f"\n {'ID':>4} {'时间':<19} {'耗时(us)':>10} {'命令':<60} {'客户端':<22} 名字")
print(" " + "-" * 130)
for sid, ts, dur, cmd, ip, nm in rows:
ts_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
print(f" {sid:>4} {ts_str:<19} {dur:>10} {cmd:<60} {ip:<22} {nm}")
print("\n ─── 优化建议 ───")
seen = set()
for sid, ts, dur, cmd, ip, nm in rows:
cmd_head = cmd.split(" ", 1)[0]
if cmd_head in seen:
continue
seen.add(cmd_head)
print(f" [{cmd_head:>10}] {suggest(cmd)}")
def main() -> None:
try:
setup_slowlog(threshold_us=1000)
trigger_slow_commands()
show_slowlog(top_n=20)
section("Step 4: 总结")
print(" ① SLOWLOG 阈值生产建议设为 1000us（1ms）")
print(" ② 关注 6 个字段：id / 时间 / 耗时 / 命令 / 客户端 IP / 客户端 name")
print(" ③ KEYS / HGETALL / LRANGE 0 -1 是最常见的「定时炸弹」")
print(" ④ 用 SCAN 系列 + UNLINK + Pipeline 改造慢命令")
except redis.ConnectionError as e:
print(f"❌ Redis 连接失败：{e}")
if __name__ == "__main__":
main()python
"""
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()python
"""
Ch13 配套代码 3 / 4 —— 内存综合分析器
演示：
1. MEMORY STATS —— 全局内存视图
2. MEMORY USAGE —— 抽样 Top N 大 Key
3. OBJECT IDLETIME —— 找出长时间未访问的「冷 Key」
4. 给出整体诊断（碎片率、Top 大 Key、冷 Key 占比）
"""
import random
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) -> 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 prepare_demo_data() -> None:
section("Step 0: 准备一批演示数据")
print(" - 100 个普通 String / 5 个大 Hash / 1 个超大 List")
pipe = r.pipeline(transaction=False)
for i in range(100):
pipe.set(f"demo:mem:str:{i}", "x" * random.randint(50, 200))
for i in range(5):
mapping = {f"f{j}": "y" * 100 for j in range(200)}
pipe.hset(f"demo:mem:hash:{i}", mapping=mapping)
pipe.rpush("demo:mem:biglist", *[f"item{i}" for i in range(5000)])
pipe.execute()
print(" ✓ 已写入约 100+5+1 个 demo Key（前缀 demo:mem:）")
def cleanup_demo_data() -> None:
keys = list(r.scan_iter(match="demo:mem:*", count=500))
if keys:
r.unlink(*keys)
def show_memory_stats() -> None:
section("Step 1: MEMORY STATS —— 全局内存视图")
stats = r.execute_command("MEMORY", "STATS")
pairs = {}
it = iter(stats)
for k in it:
v = next(it)
pairs[k] = v
interesting = [
"peak.allocated",
"total.allocated",
"startup.allocated",
"replication.backlog",
"clients.slaves",
"clients.normal",
"aof.buffer",
"lua.caches",
"overhead.total",
"keys.count",
"keys.bytes-per-key",
"dataset.bytes",
"dataset.percentage",
"allocator.allocated",
"allocator.active",
"allocator.resident",
"allocator-fragmentation.ratio",
"allocator-fragmentation.bytes",
"allocator-rss.ratio",
"rss-overhead.ratio",
"fragmentation",
]
print(f" {'指标':<32} 值")
print(" " + "-" * 60)
for k in interesting:
if k in pairs:
v = pairs[k]
if isinstance(v, (int, float)) and "ratio" not in k and "percentage" not in k and "count" not in k and "per-key" not in k:
v = fmt_bytes(v)
elif isinstance(v, float):
v = f"{v:.4f}"
print(f" {k:<32} {v}")
def show_top_keys(top_n: int = 10) -> None:
section(f"Step 2: MEMORY USAGE 抽样找 Top {top_n} 大 Key")
sampled = []
for k in r.scan_iter(match="demo:mem:*", count=500):
try:
usage = r.memory_usage(k)
if usage is not None:
sampled.append((k, usage))
except redis.ResponseError:
continue
if not sampled:
print(" （未发现 demo Key，先跑 prepare_demo_data）")
return
sampled.sort(key=lambda x: -x[1])
print(f" 共扫描 {len(sampled)} 个 Key\n")
print(f" {'Key':<40} {'类型':<10} {'内存':<12} {'IdleTime(s)':<12}")
print(" " + "-" * 80)
for k, usage in sampled[:top_n]:
try:
t = r.type(k)
idle = r.object("idletime", k) or 0
except redis.ResponseError:
t = idle = "-"
print(f" {k:<40} {t:<10} {fmt_bytes(usage):<12} {idle:<12}")
total = sum(u for _, u in sampled)
top_total = sum(u for _, u in sampled[:top_n])
print(f"\n Top {top_n} 占比：{top_total/total*100:.1f}%（{fmt_bytes(top_total)} / {fmt_bytes(total)}）")
if top_total / total > 0.5:
print(" ⚠ Top 10 个 Key 占总内存超过 50%，存在大 Key 风险！")
def show_idle_keys(threshold_sec: int = 0) -> None:
section(f"Step 3: OBJECT IDLETIME 找冷 Key（阈值 = {threshold_sec}s）")
cold = []
for k in r.scan_iter(match="demo:mem:*", count=500):
try:
idle = r.object("idletime", k) or 0
if idle >= threshold_sec:
cold.append((k, idle))
except redis.ResponseError:
continue
cold.sort(key=lambda x: -x[1])
if not cold:
print(" （没有冷 Key）")
return
print(f" 共发现冷 Key {len(cold)} 个，前 10 个：\n")
for k, idle in cold[:10]:
print(f" {k:<40} idle={idle}s")
print("\n 💡 冷 Key 处理策略：")
print(" ① 设置 TTL 让其自动过期")
print(" ② 用 UNLINK 异步删除")
print(" ③ 持久化策略改为 LRU/LFU 让其自动淘汰")
def diagnosis() -> None:
section("Step 4: 诊断结论")
info = r.info("memory")
used = int(info.get("used_memory", 0))
rss = int(info.get("used_memory_rss", 0))
frag = float(info.get("mem_fragmentation_ratio", 1.0))
print(f" used_memory = {fmt_bytes(used)}")
print(f" rss = {fmt_bytes(rss)}")
print(f" 碎片率 = {frag:.2f}")
print()
if frag > 1.5:
print(" ❌ 碎片率 > 1.5：建议开启 activedefrag 或重启从节点")
elif frag < 1.0:
print(" 🚨 碎片率 < 1.0：可能触发 SWAP！立即 free -m 检查内存")
else:
print(" ✅ 碎片率正常")
if int(info.get("evicted_keys", 0)) > 0:
print(f" ⚠ 累计淘汰 {info['evicted_keys']} 个 Key：maxmemory 设置过小")
def main() -> None:
try:
prepare_demo_data()
show_memory_stats()
show_top_keys(top_n=10)
show_idle_keys(threshold_sec=0)
diagnosis()
except redis.ConnectionError as e:
print(f"❌ Redis 连接失败：{e}")
finally:
cleanup_demo_data()
if __name__ == "__main__":
main()python
"""
Ch13 配套代码 4 / 4 —— 13 条优化清单 self-check
演示：
自动检查当前 Redis 配置 vs 13 条最佳实践
对每一条给出「✅ 通过 / ⚠ 警告 / ❌ 不通过」+ 改进建议
注意：部分条目（如客户端是否用连接池、是否用 Pipeline）只能在客户端
侧检查，本脚本会在这些条目上提示「需业务自检」。
"""
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)
PASS = "✅ PASS"
WARN = "⚠ WARN"
FAIL = "❌ FAIL"
SELF = "📋 SELF"
def cfg_get(key: str):
try:
items = r.config_get(key)
return items.get(key)
except redis.ResponseError:
return None
def parse_bytes(s: str) -> int:
if not s or s == "0":
return 0
try:
return int(s)
except ValueError:
return 0
def check_1_pool() -> tuple:
return (SELF, "客户端连接池", "客户端代码自检：使用 redis.ConnectionPool / Jedis Pool / Lettuce")
def check_2_pipeline() -> tuple:
return (SELF, "批量场景用 Pipeline / MGET / HMGET", "客户端代码自检：批量操作用 pipeline.execute()")
def check_3_lua() -> tuple:
return (SELF, "复合逻辑用 Lua 下沉", "客户端代码自检：『读-改-写』复合逻辑用 EVAL")
def check_4_ttl() -> tuple:
info = r.info("keyspace")
if not info:
return (WARN, "TTL 设置", "Keyspace 为空，无法判断")
no_ttl_dbs = []
for db, stats in info.items():
if isinstance(stats, dict):
keys = stats.get("keys", 0)
expires = stats.get("expires", 0)
ratio = expires / keys if keys else 0
if keys > 100 and ratio < 0.5:
no_ttl_dbs.append((db, keys, expires, ratio))
if no_ttl_dbs:
msg = "； ".join(f"{db} TTL 覆盖率 {r*100:.0f}%" for db,_,_,r in no_ttl_dbs)
return (WARN, "TTL 设置", f"以下 db 大量 Key 无 TTL：{msg}")
return (PASS, "TTL 设置", "Key 普遍设置了 TTL")
def check_5_listpack() -> tuple:
e = int(cfg_get("hash-max-listpack-entries") or 128)
v = int(cfg_get("hash-max-listpack-value") or 64)
if e <= 128 and v <= 64:
return (PASS, "listpack 阈值合理",
f"hash-max-listpack-entries={e}, hash-max-listpack-value={v}")
return (WARN, "listpack 阈值偏大",
f"entries={e}, value={v}（建议 ≤128 / ≤64，让小 Hash 用紧凑编码）")
def check_6_keylen() -> tuple:
sample = []
for k in r.scan_iter(count=500):
sample.append(len(k))
if len(sample) >= 1000:
break
if not sample:
return (WARN, "Key 长度", "未发现 Key，无法采样")
avg = sum(sample) / len(sample)
mx = max(sample)
if avg > 50:
return (WARN, "Key 长度", f"平均 Key 长度 {avg:.0f}（建议 < 50），最大 {mx}")
return (PASS, "Key 长度", f"平均 Key 长度 {avg:.0f}，最大 {mx}")
def check_7_dangerous_cmds() -> tuple:
danger = ["KEYS", "FLUSHALL", "FLUSHDB", "DEBUG", "CONFIG"]
enabled = []
for cmd in danger:
try:
res = r.execute_command("COMMAND", "INFO", cmd)
if res and res[0] is not None:
enabled.append(cmd)
except redis.ResponseError:
continue
if enabled:
return (WARN, "危险命令未禁用",
f"以下命令仍可用：{', '.join(enabled)}（生产建议 rename-command 禁用）")
return (PASS, "危险命令已禁用", "KEYS/FLUSHALL/DEBUG 等已禁用")
def check_8_unlink_lazy() -> tuple:
flags = {
"lazyfree-lazy-eviction": cfg_get("lazyfree-lazy-eviction"),
"lazyfree-lazy-expire": cfg_get("lazyfree-lazy-expire"),
"lazyfree-lazy-server-del": cfg_get("lazyfree-lazy-server-del"),
"lazyfree-lazy-user-del": cfg_get("lazyfree-lazy-user-del"),
}
off = [k for k, v in flags.items() if (v or "no").lower() != "yes"]
if off:
return (WARN, "lazy free 未全开",
f"未开启：{', '.join(off)}（建议全设 yes，让删除走 BIO 异步）")
return (PASS, "lazy free 全开", "DEL/EXPIRE/淘汰 都异步释放")
def check_9_bigkey() -> tuple:
big = []
for k in r.scan_iter(count=500):
try:
usage = r.memory_usage(k)
if usage and usage > 100 * 1024:
big.append((k, usage))
except redis.ResponseError:
continue
if len(big) >= 5:
break
if big:
msg = "； ".join(f"{k}={u//1024}KB" for k, u in big[:3])
return (FAIL, "存在大 Key", f"≥100KB 的 Key 抽样：{msg}（应拆分）")
return (PASS, "大 Key 检查", "抽样未发现 ≥100KB 的 Key")
def check_10_isolation() -> tuple:
info = r.info("keyspace")
db_count = len([k for k in info if k.startswith("db")])
if db_count > 4:
return (WARN, "业务隔离",
f"使用了 {db_count} 个 DB，混用业务的可能性高（建议拆实例而非用多 DB）")
return (SELF, "业务隔离",
"需架构层面确认：不同业务是否使用了独立 Redis 实例")
def check_11_rdb_save() -> tuple:
save = cfg_get("save")
if save and save.strip():
return (WARN, "RDB 自动保存",
f"save = '{save}'。生产环境建议关闭自动 RDB，改 cron 错峰触发 BGSAVE")
return (PASS, "RDB 自动保存", "已禁用，可走 cron 错峰触发")
def check_12_maxmemory() -> tuple:
mm = parse_bytes(cfg_get("maxmemory") or "0")
info = r.info("memory")
used = int(info.get("used_memory", 0))
if mm == 0:
return (FAIL, "maxmemory 未限制",
"maxmemory=0 危险！请设置上限（建议留 30% 内存给 fork）")
if used / mm > 0.7:
return (WARN, "maxmemory 利用率高",
f"已用 {used/mm*100:.0f}% / {mm//1024//1024}MB（>70% 进入告警区）")
return (PASS, "maxmemory 设置合理",
f"已用 {used/mm*100:.0f}% / {mm//1024//1024}MB")
def check_13_linux() -> tuple:
notes = []
try:
with open("/proc/sys/vm/overcommit_memory") as f:
v = f.read().strip()
if v != "1":
notes.append(f"overcommit_memory={v}（建议 1）")
except OSError:
notes.append("无法读取 /proc/sys/vm/overcommit_memory")
try:
with open("/sys/kernel/mm/transparent_hugepage/enabled") as f:
v = f.read().strip()
if "[never]" not in v:
notes.append(f"THP={v}（建议 never）")
except OSError:
notes.append("无法读取 transparent_hugepage")
try:
with open("/proc/sys/vm/swappiness") as f:
v = int(f.read().strip())
if v > 10:
notes.append(f"swappiness={v}（建议 ≤10）")
except OSError:
notes.append("无法读取 swappiness")
if notes:
return (WARN, "Linux 内核参数", "； ".join(notes))
return (PASS, "Linux 内核参数", "overcommit/THP/swappiness 配置 OK")
CHECKS = [
("1. 连接池", check_1_pool),
("2. Pipeline / MGET", check_2_pipeline),
("3. Lua 下沉", check_3_lua),
("4. TTL 设置", check_4_ttl),
("5. listpack 阈值", check_5_listpack),
("6. Key 长度", check_6_keylen),
("7. 禁用危险命令", check_7_dangerous_cmds),
("8. UNLINK / lazy free", check_8_unlink_lazy),
("9. 大 Key 检查", check_9_bigkey),
("10. 业务隔离", check_10_isolation),
("11. 错峰主从同步", check_11_rdb_save),
("12. maxmemory 留余量", check_12_maxmemory),
("13. Linux 内核调优", check_13_linux),
]
def main() -> None:
try:
section("Redis 13 条优化清单 self-check")
results = []
for name, fn in CHECKS:
try:
status, _, msg = fn()
except Exception as e:
status, msg = FAIL, f"check error: {e}"
results.append((name, status, msg))
print(f"\n [{status}] {name}")
print(f" {msg}")
section("最终诊断报告")
n_pass = sum(1 for _, s, _ in results if s == PASS)
n_warn = sum(1 for _, s, _ in results if s == WARN)
n_fail = sum(1 for _, s, _ in results if s == FAIL)
n_self = sum(1 for _, s, _ in results if s == SELF)
print(f" ✅ 通过：{n_pass} ⚠ 警告：{n_warn} ❌ 不通过：{n_fail} 📋 需自检：{n_self}")
score = (n_pass + n_self * 0.5) / len(results) * 100
print(f" 自动评分：{score:.0f} / 100\n")
critical = [n for n, s, _ in results if s == FAIL]
if critical:
print(" 🚨 必须立即处理：")
for n in critical:
print(f" - {n}")
warn_items = [n for n, s, _ in results if s == WARN]
if warn_items:
print("\n ⚠ 建议改进：")
for n in warn_items:
print(f" - {n}")
if not critical and not warn_items:
print(" 🎉 全部通过，配置堪称完美！")
except redis.ConnectionError as e:
print(f"❌ Redis 连接失败：{e}")
if __name__ == "__main__":
main()01_slowlog_analyzer.py ↗ · 02_info_dashboard.py ↗ · 03_memory_analyzer.py ↗ · 04_optimization_checklist.py ↗