主题
第 4 章 Key 设计与命令进阶
学习目标:写出优雅、可维护、不踩坑的 Redis Key;掌握 SCAN 系列命令安全遍历海量数据;理解 EXPIRE/TTL 的细节与陷阱;学会用 OBJECT/MEMORY 命令排查问题。
4.1 Key 命名规范
Key 是字符串,看似随便起个名就行——但规范的命名能在生产环境救你的命。
4.1.1 七大原则
✅ 推荐 ❌ 反面教材
─────────────────────────────────────────────────
1. 用「冒号」分层 user:1001:profile vs user_1001_profile
2. 用业务/对象/字段三段式 order:202604:item:001 vs o_x_y
3. 控制长度(< 44 字节最佳) u:1001 vs user_profile_full_data_for_id_1001
4. 不用空格、特殊字符 mylist vs "my list!"
5. 二进制安全但可读为先 cart:user:1 vs \x01\x02\x03
6. 加上版本号便于灰度 v2:product:123 vs product:123(升级时痛苦)
7. 加业务前缀防止冲突 shop:order:1, mall:order:14.1.2 推荐的 Key 模板
{业务前缀}:{对象名}:{对象 ID}[:{字段名}]
─────────────────────────────────────────────
shop:user:1001 # 用户对象(Hash)
shop:user:1001:cart # 用户购物车(Hash 或 ZSet)
shop:order:202604170001 # 订单
shop:rank:product:daily # 日销量榜(ZSet)
shop:lock:order:1001 # 分布式锁
shop:limit:user:1001:download # 限流计数(带过期)变长字段优先放最后——如果带 : 或动态拼接,前面的部分要稳定。
4.1.3 长度对内存的影响
bash
# 同样的功能,Key 名长 vs 短
SET this:is:a:very:long:key:name:for:user:profile:1001 "data"
SET u:1001 "data"
# 用 MEMORY USAGE 测一下
MEMORY USAGE this:is:a:very:long:key:name:for:user:profile:1001 # 130 字节
MEMORY USAGE u:1001 # 64 字节
↑ 节省 50%!百万个 Key 累计起来差距巨大。短 Key 名就是省钱。
4.2 通用命令总览
bash
# ===== 存在 / 删除 / 类型 =====
EXISTS key1 key2 # 返回存在的个数
DEL key1 key2 # 同步删除
UNLINK key1 key2 # 异步删除(大 Key 必用!)
TYPE key # string / list / hash / set / zset / stream / none
# ===== 查找(小心!)=====
KEYS pattern # ⚠️ 阻塞,禁用!
SCAN 0 MATCH pattern COUNT 100 # ✅ 渐进式遍历
# ===== 改名 / 复制 / 移动 =====
RENAME k1 k2 # 改名(原子)
RENAMENX k1 k2 # 仅当 k2 不存在时改
COPY src dst REPLACE # 复制(6.2+)
MOVE key 1 # 把 key 移到 1 号 db(不推荐用多 db)
# ===== 过期 =====
EXPIRE key 60 # 60 秒后过期
PEXPIRE key 60000 # 毫秒
EXPIREAT key 1719720000 # Unix 时间戳
TTL key # 还剩多少秒(-1=永不过期,-2=不存在)
PTTL key # 毫秒级
PERSIST key # 移除过期,变成永久
# ===== 元信息 =====
OBJECT ENCODING key # 底层编码
OBJECT FREQ key # 访问频率(LFU 策略时有效)
OBJECT IDLETIME key # 空闲时长(秒)
MEMORY USAGE key # 占用字节
DEBUG OBJECT key # 详细调试信息(生产慎用)4.3 SCAN:海量 Key 的安全遍历
4.3.1 为什么不能用 KEYS *?
Redis 是单线程的!
KEYS pattern 的工作方式:
for each key in 全量字典:
if key matches pattern:
收集到结果列表
return 结果列表
→ 1000 万 Key 时可能阻塞主线程数秒!
→ 这期间所有客户端请求被打回
→ 上游服务雪崩
真实案例:
某团队凌晨 KEYS user:* → 主线程阻塞 4 秒
→ 200 台 Web 机的连接全部超时
→ 业务雪崩 5 分钟结论:KEYS 命令在生产环境必须禁用。可以在 redis.conf 用 rename-command KEYS "" 直接屏蔽。
4.3.2 SCAN 怎么用
bash
# 第一次:cursor = 0
SCAN 0 MATCH "user:*" COUNT 100
# 1) "1024" ← 下次的 cursor
# 2) 1) "user:1"
# 2) "user:5"
# ...
# 用上次返回的 cursor 继续
SCAN 1024 MATCH "user:*" COUNT 100
# 1) "2048"
# 2) (...)
# 反复直到 cursor 变回 "0",表示遍历结束关键参数:
MATCH pattern:客户端过滤(不是服务端,所以可能有几次 SCAN 返回空数组)COUNT N:期望每次返回的元素数(不是精确,只是 hint)TYPE type:仅返回指定类型的 Key(6.0+)
伪代码版的安全遍历:
python
import redis
r = redis.Redis()
cursor = 0
while True:
cursor, keys = r.scan(cursor, match="user:*", count=200)
for key in keys:
process(key)
if cursor == 0:
break4.3.3 SCAN 的「保证」与「不保证」
✅ 保证:
- 单次 SCAN 调用是 O(1) 的(小批次)
- 整个遍历过程是 O(N),分摊到多次调用,不阻塞主线程
- 「遍历开始时就存在的 Key」一定会被返回
❌ 不保证:
- 每次返回的元素数恰好等于 COUNT(COUNT 只是 hint)
- 遍历过程中**新增**的 Key 不一定被返回
- 同一个 Key 可能被**返回多次**!应用层要去重为什么可能重复? SCAN 用的是「反向二进制迭代」算法,避免漏掉,但代价是可能重复。这是 SCAN 设计上的权衡。
4.3.4 SCAN 系列:HSCAN / SSCAN / ZSCAN
bash
# Hash 的所有字段(避免 HGETALL 阻塞)
HSCAN user:1001 0 COUNT 50
# Set 的所有成员
SSCAN tags 0 MATCH "redis*" COUNT 100
# ZSet 的所有成员(带 score)
ZSCAN rank 0 COUNT 100何时必用? 当你怀疑 Hash/Set/ZSet 是大 Key 时,用 HSCAN/SSCAN/ZSCAN 替代 HGETALL/SMEMBERS/ZRANGE 0 -1。
4.4 过期机制:精妙的「定时 + 惰性 + 定期」
💡 这是 Ch5「过期与内存淘汰」的预热,本节先讲 EXPIRE 的使用与陷阱,机制详解放到 Ch5。
4.4.1 EXPIRE 的常见操作
bash
SET token "abc" EX 3600 # SET 时直接带过期(推荐)
SET token "abc" PX 3600000 # 毫秒版
SETEX token 3600 "abc" # 等价上面(老写法)
PSETEX token 3600000 "abc" # 毫秒版老写法
EXPIRE token 60 # 给已存在的 Key 设过期
EXPIREAT token 1719720000 # Unix 时间戳(绝对时间)
TTL token # 还剩多少秒
PERSIST token # 移除过期,变成永久
# 7.0+ 新增的可选项
EXPIRE token 60 NX # 仅当 Key 当前没有 TTL 时才设置
EXPIRE token 60 XX # 仅当 Key 当前有 TTL 时才设置
EXPIRE token 60 GT # 仅当新 TTL > 旧 TTL 时设置
EXPIRE token 60 LT # 仅当新 TTL < 旧 TTL 时设置4.4.2 易错点 ⚠️
坑 1:SET key value 会清除原有的过期时间
bash
SET token "v1" EX 60 # TTL = 60s
TTL token # → 60
SET token "v2" # 没带 EX!
TTL token # → -1 ← 过期被清掉了!正确做法:每次 SET 都带 EX 或 KEEPTTL(6.0+):
bash
SET token "v2" KEEPTTL # 保留原有 TTL坑 2:RENAME 会继承源 Key 的 TTL
bash
SET k1 "v1" EX 60
SET k2 "v2" # k2 没有过期
RENAME k1 k2 # k2 现在 TTL=60(继承自 k1)坑 3:负数或 0 的 EXPIRE = 立即删除
bash
SET k "v"
EXPIRE k 0 # 等同于 DEL k
EXPIRE k -1 # 也是立即删坑 4:业务要的是「绝对时间」
bash
# ❌ 用 EXPIRE 60,进程睡眠/重启会推迟过期
EXPIRE token 60
# ✅ 用 EXPIREAT,无论何时都精准
EXPIREAT token $(date -d "+1 hour" +%s)4.5 大 Key / 热 Key 排查
4.5.1 什么是大 Key?
| 类型 | 大 Key 警戒线 |
|---|---|
| String | > 10 KB(极限是 512 MB) |
| List / Set / Hash / ZSet | 元素数 > 5000 |
| 单个 Key 总内存 | > 1 MB |
大 Key 的危害:
- DEL 大 Key 阻塞主线程(→ 用 UNLINK 异步删除)
- 集群下导致数据倾斜
- 网络传输打爆带宽
- 持久化 fork 时 COW 写时复制成本剧增
4.5.2 排查工具
bash
# 1. 客户端工具:bigkeys(最常用)
redis-cli --bigkeys
# 扫描整个 db,输出每种类型最大的 Key
# 用的是 SCAN,不会阻塞,但慢
# 2. memkeys(同 bigkeys 但按内存排序)
redis-cli --memkeys
# 3. 单 Key 排查
MEMORY USAGE bigkey # 占用字节数
DEBUG OBJECT bigkey # 详细信息(含编码、引用数)
# 4. 离线分析 RDB 文件(推荐)
# 工具:rdb-tools(pip install rdbtools)
rdb -c memory dump.rdb --bytes 10240 > big_keys.csv4.5.3 大 Key 治理思路
1. 拆分:1 个超大 Hash → 100 个小 Hash
user:profile (10MB) → user:profile:0..99 (各 100KB)
按 hash(field) % 100 分桶
2. 异步删除:
❌ DEL big_key # 阻塞
✅ UNLINK big_key # 后台线程删除(4.0+)
3. 替换数据结构:
- 大 Hash 字段全是数字 → 用 Bitmap
- 大 Set 用于精确去重 → 评估改用 HyperLogLog
4. 分层缓存:
- 大 Key 不放 Redis,放本地缓存或专用存储4.5.4 热 Key 排查
bash
# 4.0+ 内置 hotkeys(要先开启 LFU)
config set maxmemory-policy allkeys-lfu
redis-cli --hotkeys
# 通过 MONITOR 抓取实时命令(仅开发环境,生产严禁!)
MONITOR
# 1719720000.000 [0 1.2.3.4:50000] "GET" "hot_key"
# ...
# 通过外部工具:facebook 开源的 redis-faina 分析
redis-cli MONITOR | head -n 100000 | redis-faina.py热 Key 治理:
- 多副本:将热点写到多个 Key(
hot_key:0、hot_key:1...),客户端轮询读 - 本地缓存:JVM/进程内缓存承担大头(注意一致性)
- 读写分离:读走从节点
4.6 实操:跑一遍配套代码
实战代码见 04_key_design/code/:
01_scan_vs_keys.py:百万 Key 场景下 KEYS vs SCAN 的耗时对比02_expire_pitfalls.py:演示 SET 清除 TTL、RENAME 继承 TTL 等坑03_bigkey_finder.py:手写一个简易 bigkeys 扫描器
浏览器演示见 04_key_design/demo.html:
- ① 渐进式 SCAN 动画(看游标怎么走)
- ② Key 命名打分器(输入 Key 名,给出评分和建议)
- ③ TTL 倒计时可视化
4.7 本章小结
┌──────────────────────────────────────────────────────────┐
│ 本章核心要点 │
├──────────────────────────────────────────────────────────┤
│ │
│ ① Key 命名:{业务}:{对象}:{ID}[:{字段}],短而清晰 │
│ │
│ ② KEYS / FLUSHALL / FLUSHDB 是生产禁用命令 │
│ 用 SCAN 渐进式遍历,HSCAN/SSCAN/ZSCAN 同理 │
│ │
│ ③ SET 默认会清除 TTL!要保留就用 SET ... KEEPTTL │
│ │
│ ④ 大 Key 排查:redis-cli --bigkeys │
│ 治理:拆分 / UNLINK 异步删 / 换数据结构 │
│ │
│ ⑤ 热 Key 排查:--hotkeys(需 LFU) │
│ 治理:多副本 / 本地缓存 / 读写分离 │
│ │
└──────────────────────────────────────────────────────────┘4.8 面试高频题
Q1:为什么生产环境禁用 KEYS *?应该用什么替代?
考察点:基础常识 + 解决方案。
标准答案:
KEYS pattern 会遍历整个键空间,由于 Redis 是单线程的,这期间会阻塞所有其他命令。在 Key 数量上百万级时可能阻塞数秒到数十秒,导致:
- 客户端连接超时;
- 上游服务雪崩;
- 哨兵误判主节点宕机触发故障转移。
替代方案:用 SCAN 渐进式遍历。每次只扫一小批,配合 cursor 多次调用直到回到 0。SCAN 是 O(1) 单次调用 + O(N) 总耗时分摊。
生产配置:在 redis.conf 中用 rename-command KEYS "" 直接屏蔽该命令。
Q2:SCAN 命令有哪些「不保证」?怎么处理?
考察点:对 SCAN 局限性的理解。
标准答案:
SCAN 不保证:
- 每次返回的元素数恰好等于 COUNT(COUNT 只是 hint);
- 遍历过程中新增的 Key 不一定被返回;
- 同一个 Key 可能被返回多次(反向二进制迭代算法的代价)。
应用层处理:
- 对结果做 去重(用 Set 存已处理的 Key);
- 对「必须不漏」的场景,先停止写入再扫描,或用其他方案(如订阅 keyspace 事件)。
保证:遍历开始时就存在的 Key 一定会被返回(不漏旧数据)。
Q3:DEL 和 UNLINK 有什么区别?什么时候必须用 UNLINK?
考察点:大 Key 治理。
标准答案:
| 命令 | 工作方式 | 适用场景 |
|---|---|---|
| DEL | 主线程同步删除,立即释放内存 | 小 Key |
| UNLINK | 主线程只解除 Key→Object 的映射,释放内存的工作交给 BIO 后台线程(4.0+) | 大 Key |
对比:
- 删除一个 1000 万元素的 List,
DEL会阻塞主线程数百毫秒; - 同样数据用
UNLINK,主线程只阻塞微秒级,真正的内存释放在后台异步完成。
最佳实践:生产环境一律用 UNLINK(小 Key 用 UNLINK 也只是多一次跨线程调度,几乎无开销)。
Q4:Redis 怎么删除过期 Key?为什么会有「定期 + 惰性 + 定时」三种?
考察点:过期机制(Ch5 详细讲,这里答个轮廓)。
标准答案:
Redis 用惰性删除 + 定期删除结合的策略:
惰性删除(Lazy):访问 Key 时检查是否过期,过期就删。优点:CPU 友好;缺点:从不被访问的过期 Key 一直占用内存。
定期删除(Active):每秒 10 次(默认 hz=10),随机抽样 20 个有过期时间的 Key 检查并清理过期的。如果过期比例 > 25%,重复扫描直到比例下降。优点:清理冷数据;缺点:CPU 和内存的折中。
过期事件通知:
notify-keyspace-events Ex开启后,过期时会发布事件到__keyevent@0__:expired频道(不是「主动删除策略」,是机制配套)。
注意:不存在「定时删除」(每个 Key 都创建定时器太重),网上很多文章说错了。
Q5:SET 命令会改变过期时间吗?
考察点:易错细节。
标准答案:
会! 默认的 SET key value 会清除该 Key 的过期时间,变成永久。
bash
SET k "v1" EX 60 # TTL = 60
SET k "v2" # TTL 变成 -1(永久)避免方案:
- 6.0+ 用
SET key value KEEPTTL保留原有 TTL; - 老版本要么
SET + 立即 EXPIRE(非原子),要么用 Lua 脚本保证原子。
易错衍生:
RENAME会继承源 Key 的 TTL;INCR/APPEND等修改命令不会改变 TTL;EXPIRE k 0或负数 = 立即删除。
Q6:怎么发现并解决大 Key 问题?
考察点:综合应用 + 工程经验。
标准答案:
发现:
- redis-cli --bigkeys:在线扫描,每种类型最大的 Key(用 SCAN 实现,不阻塞);
- redis-cli --memkeys:按字节数排序;
- rdb-tools 离线分析:分析 RDB 文件,得到完整 Key 大小分布;
- MEMORY USAGE key:单 Key 精确测量;
- 客户端慢日志:结合
SLOWLOG看是否有大 Key 操作。
解决:
- 拆分:大 Hash 按字段 hash 分桶到多个小 Hash;大 List 按时间或 ID 分片。
- 删除:必须用
UNLINK异步删除,禁用DEL。 - 改结构:
- Hash 字段全是布尔 → Bitmap;
- Set 用于近似去重 → HyperLogLog;
- 大 ZSet 排行榜 → 按分数段拆分多个小 ZSet。
- 业务侧:评估「是否真的需要这么大的 Key」,往往是设计冗余。
预防:
- 上线前做 Key 设计评审;
- 在客户端 SDK 加大小检测(写入前预估);
- 监控告警:单 Key 大小超阈值自动报警。
📌 下一章预告:第 5 章我们深入「过期与内存淘汰」——Redis 如何在内存吃紧时自动清理数据?8 种淘汰策略怎么选?LRU 和 LFU 的近似实现是什么?
🎬 可视化演示
演示加载缓慢或样式异常?点此在新标签页打开 ↗
💻 示例代码
python
"""
Ch4 配套代码 1 / 3 —— SCAN vs KEYS 性能对比
- 灌入 5 万个 Key
- 分别用 KEYS 和 SCAN 遍历,对比单次阻塞时间
- SCAN 总耗时可能更长,但单次极短,不会阻塞主线程
"""
import time
import redis
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
PREFIX = "demo:scan:"
def section(t): print("\n" + "=" * 60 + f"\n{t}\n" + "=" * 60)
def setup(n: int) -> None:
print(f"[setup] 灌入 {n} 个 Key ...")
pipe = r.pipeline(transaction=False)
for i in range(n):
pipe.set(f"{PREFIX}{i}", "v")
if i % 1000 == 999: pipe.execute()
pipe.execute()
def cleanup() -> None:
cursor = 0
while True:
cursor, keys = r.scan(cursor, match=f"{PREFIX}*", count=1000)
if keys: r.delete(*keys)
if cursor == 0: break
def use_keys() -> None:
section("方案 A: KEYS(单次阻塞)⚠️ 生产禁用")
start = time.perf_counter()
keys = r.keys(f"{PREFIX}*")
cost = (time.perf_counter() - start) * 1000
print(f" 返回 {len(keys)} 个 Key")
print(f" ⚠️ 单次调用耗时 {cost:.2f} ms(这段时间主线程阻塞,所有客户端等待)")
def use_scan() -> None:
section("方案 B: SCAN(多次小调用)✅ 安全")
cursor = 0
rounds = 0
max_call_ms = 0
total_keys = 0
start = time.perf_counter()
while True:
s = time.perf_counter()
cursor, keys = r.scan(cursor, match=f"{PREFIX}*", count=500)
single = (time.perf_counter() - s) * 1000
max_call_ms = max(max_call_ms, single)
total_keys += len(keys)
rounds += 1
if cursor == 0: break
total = (time.perf_counter() - start) * 1000
print(f" 调用 {rounds} 次,累计返回 {total_keys} 个 Key(含可能的重复)")
print(f" 总耗时 {total:.2f} ms(分摊到多次,主线程不阻塞)")
print(f" ✓ 单次最大耗时 {max_call_ms:.2f} ms(安全)")
if __name__ == "__main__":
try:
cleanup()
setup(50000)
use_keys()
use_scan()
finally:
cleanup()python
"""
Ch4 配套代码 2 / 3 —— EXPIRE 的常见坑实测
坑 1: SET 默认会清除 TTL
坑 2: RENAME 会继承源 Key 的 TTL
坑 3: EXPIRE 0 / 负数 = 立即删除
坑 4: KEEPTTL 的正确用法
"""
import time
import redis
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def section(t): print("\n" + "=" * 60 + f"\n{t}\n" + "=" * 60)
def pitfall_1_set_clears_ttl() -> None:
section("坑 1: SET 默认会清除 TTL")
r.set("token", "v1", ex=60)
print(f" SET token 'v1' EX 60 → TTL = {r.ttl('token')}")
r.set("token", "v2") # 不带 EX
print(f" SET token 'v2' (无 EX) → TTL = {r.ttl('token')} ❌ 过期被清掉了")
r.delete("token")
def pitfall_2_rename_inherits_ttl() -> None:
section("坑 2: RENAME 会继承源 Key 的 TTL")
r.set("k1", "v1", ex=120)
r.set("k2", "v2") # k2 没有过期
print(f" k1 TTL = {r.ttl('k1')}, k2 TTL = {r.ttl('k2')}")
r.rename("k1", "k2")
print(f" RENAME k1 k2 之后:")
print(f" k2 TTL = {r.ttl('k2')} ← 继承自 k1")
r.delete("k2")
def pitfall_3_expire_zero() -> None:
section("坑 3: EXPIRE 0/负数 = 立即删除")
r.set("k", "v")
print(f" EXISTS 之前: {r.exists('k')}")
r.expire("k", 0)
print(f" EXPIRE k 0 之后 EXISTS: {r.exists('k')} ← 立即被删")
def fix_keep_ttl() -> None:
section("正确用法: SET ... KEEPTTL(6.0+)")
r.set("token", "v1", ex=60)
print(f" SET ... EX 60 → TTL = {r.ttl('token')}")
r.set("token", "v2", keepttl=True)
print(f" SET ... KEEPTTL → TTL = {r.ttl('token')} ✅ 保留")
r.delete("token")
def demo_expireat_vs_expire() -> None:
section("EXPIRE vs EXPIREAT:相对时间 vs 绝对时间")
r.set("k1", "v", ex=10)
r.expireat("k2", int(time.time()) + 10)
r.set("k2", "v") # 注意:SET 又会清掉 TTL!
r.expireat("k2", int(time.time()) + 10)
print(f" EXPIRE k1 10 → TTL = {r.ttl('k1')}")
print(f" EXPIREAT k2 +10s → TTL = {r.ttl('k2')}")
print(" 💡 进程睡眠/重启时,EXPIREAT 不受影响,EXPIRE 会被推迟")
r.delete("k1", "k2")
if __name__ == "__main__":
try:
pitfall_1_set_clears_ttl()
pitfall_2_rename_inherits_ttl()
pitfall_3_expire_zero()
fix_keep_ttl()
demo_expireat_vs_expire()
except redis.ConnectionError as e:
print(f"❌ Redis 连接失败: {e}")python
"""
Ch4 配套代码 3 / 3 —— 手写一个简易 bigkeys 扫描器
- 用 SCAN 渐进式遍历所有 Key
- 对每种类型分别记录最大的 N 个
- 输出汇总报告,定位需要治理的 Key
"""
import heapq
import redis
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
TOP_N = 5
SCAN_COUNT = 200
def section(t): print("\n" + "=" * 60 + f"\n{t}\n" + "=" * 60)
def gen_demo_data() -> None:
"""造一些大小不一的 Key,方便看效果。"""
section("[setup] 造测试数据:String/List/Hash/Set/ZSet 各 5 个,大小不一")
for i in range(5):
r.set(f"demo:str:{i}", "x" * (10 ** (i + 1)))
r.delete(f"demo:list:{i}")
if i > 0: r.rpush(f"demo:list:{i}", *[f"v{j}" for j in range(10 * (i + 1))])
r.delete(f"demo:hash:{i}")
if i > 0: r.hset(f"demo:hash:{i}", mapping={f"f{j}": f"v{j}" for j in range(20 * (i + 1))})
r.delete(f"demo:set:{i}")
if i > 0: r.sadd(f"demo:set:{i}", *[f"m{j}" for j in range(15 * (i + 1))])
r.delete(f"demo:zset:{i}")
if i > 0: r.zadd(f"demo:zset:{i}", {f"m{j}": j for j in range(25 * (i + 1))})
def cleanup() -> None:
cursor = 0
while True:
cursor, keys = r.scan(cursor, match="demo:*", count=200)
if keys: r.delete(*keys)
if cursor == 0: break
def find_big_keys() -> None:
section("[scan] 扫描 demo:* 找出每种类型 Top 5 大 Key")
# 每种类型一个最小堆,存 (size, key)
heaps = {t: [] for t in ["string", "list", "hash", "set", "zset"]}
type_handlers = {
"string": lambda k: r.strlen(k),
"list": lambda k: r.llen(k),
"hash": lambda k: r.hlen(k),
"set": lambda k: r.scard(k),
"zset": lambda k: r.zcard(k),
}
cursor = 0
total_scanned = 0
while True:
cursor, keys = r.scan(cursor, match="demo:*", count=SCAN_COUNT)
for k in keys:
t = r.type(k)
if t not in type_handlers: continue
try:
size = type_handlers[t](k)
heap = heaps[t]
if len(heap) < TOP_N:
heapq.heappush(heap, (size, k))
elif size > heap[0][0]:
heapq.heapreplace(heap, (size, k))
total_scanned += 1
except redis.RedisError:
continue
if cursor == 0: break
print(f" 共扫描 {total_scanned} 个 Key\n")
for t, heap in heaps.items():
if not heap: continue
print(f" 📦 {t} 类型 Top {len(heap)}:")
for size, k in sorted(heap, reverse=True):
mem = r.memory_usage(k) or 0
unit = "字节" if t == "string" else "元素"
print(f" {k:<25} {size:>6} {unit} 内存约 {mem} 字节")
print()
if __name__ == "__main__":
try:
cleanup()
gen_demo_data()
find_big_keys()
finally:
cleanup()01_scan_vs_keys.py ↗ · 02_expire_pitfalls.py ↗ · 03_bigkey_finder.py ↗