Skip to content

第 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:1

4.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.confrename-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:
        break

4.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 都带 EXKEEPTTL(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.csv

4.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:0hot_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 不保证:

  1. 每次返回的元素数恰好等于 COUNT(COUNT 只是 hint);
  2. 遍历过程中新增的 Key 不一定被返回
  3. 同一个 Key 可能被返回多次(反向二进制迭代算法的代价)。

应用层处理

  • 对结果做 去重(用 Set 存已处理的 Key);
  • 对「必须不漏」的场景,先停止写入再扫描,或用其他方案(如订阅 keyspace 事件)。

保证:遍历开始时就存在的 Key 一定会被返回(不漏旧数据)。


考察点:大 Key 治理。

标准答案

命令工作方式适用场景
DEL主线程同步删除,立即释放内存小 Key
UNLINK主线程只解除 Key→Object 的映射,释放内存的工作交给 BIO 后台线程(4.0+)大 Key

对比

  • 删除一个 1000 万元素的 List,DEL 会阻塞主线程数百毫秒;
  • 同样数据用 UNLINK,主线程只阻塞微秒级,真正的内存释放在后台异步完成。

最佳实践生产环境一律用 UNLINK(小 Key 用 UNLINK 也只是多一次跨线程调度,几乎无开销)。


Q4:Redis 怎么删除过期 Key?为什么会有「定期 + 惰性 + 定时」三种?

考察点:过期机制(Ch5 详细讲,这里答个轮廓)。

标准答案

Redis 用惰性删除 + 定期删除结合的策略:

  1. 惰性删除(Lazy):访问 Key 时检查是否过期,过期就删。优点:CPU 友好;缺点:从不被访问的过期 Key 一直占用内存。

  2. 定期删除(Active):每秒 10 次(默认 hz=10),随机抽样 20 个有过期时间的 Key 检查并清理过期的。如果过期比例 > 25%,重复扫描直到比例下降。优点:清理冷数据;缺点:CPU 和内存的折中。

  3. 过期事件通知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 问题?

考察点:综合应用 + 工程经验。

标准答案

发现

  1. redis-cli --bigkeys:在线扫描,每种类型最大的 Key(用 SCAN 实现,不阻塞);
  2. redis-cli --memkeys:按字节数排序;
  3. rdb-tools 离线分析:分析 RDB 文件,得到完整 Key 大小分布;
  4. MEMORY USAGE key:单 Key 精确测量;
  5. 客户端慢日志:结合 SLOWLOG 看是否有大 Key 操作。

解决

  1. 拆分:大 Hash 按字段 hash 分桶到多个小 Hash;大 List 按时间或 ID 分片。
  2. 删除:必须用 UNLINK 异步删除,禁用 DEL
  3. 改结构
    • Hash 字段全是布尔 → Bitmap;
    • Set 用于近似去重 → HyperLogLog;
    • 大 ZSet 排行榜 → 按分数段拆分多个小 ZSet。
  4. 业务侧:评估「是否真的需要这么大的 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 ↗