第 2 章五大基础数据类型

学习目标：彻底搞清楚 Redis 五大基础类型（String / List / Hash / Set / ZSet）的命令、底层编码、编码切换条件、典型应用场景。看完之后能解释「为什么 ZSet 用跳表而不用红黑树」「Hash 什么时候从 listpack 升级为 hashtable」这类追问。

2.0 概览：5 种类型 × N 种编码

Redis 暴露给用户的类型只有 5 种，但每种类型在内部根据数据规模和元素类型，会自动选择不同的底层编码——这是 Redis 「内存友好 + 速度快」的关键设计。

┌────────────┬─────────────────────────────────────────────────────┐
│  Type      │  底层编码（Encoding）                                │
├────────────┼─────────────────────────────────────────────────────┤
│  String    │  int  ┃  embstr  ┃  raw                              │
│  List      │  quicklist （内部由若干 listpack 节点组成）            │
│  Hash      │  listpack  ┃  hashtable                              │
│  Set       │  intset  ┃  listpack  ┃  hashtable                   │
│  ZSet      │  listpack  ┃  skiplist + hashtable                   │
└────────────┴─────────────────────────────────────────────────────┘

   ⚙️ 编码切换原则：
   小数据 → 用紧凑编码（节省内存，O(N) 但 N 很小所以快）
   大数据 → 用专业结构（内存换速度，O(1) 或 O(log N)）

💡 用 OBJECT ENCODING <key> 命令可以查看任意 Key 的当前编码。

2.1 String：最简单也最复杂

2.1.1 是什么 + 生活类比

String 就是一个 KV 中的 V，可以存：

普通文本："Hello"
数字（自动当整数处理）："100"
二进制数据：图片字节、序列化对象

🍱 生活类比：每个 String 就像便利店的一格储物柜，柜子里放什么都行——纸条、零食、手机。柜门只有「锁/开」两个动作（SET/GET），但里面塞什么自由发挥。

2.1.2 常用命令

bash

# 基本读写
SET mykey "hello"
GET mykey                          # → "hello"
SETEX mykey 60 "hello"             # SET + 过期 60 秒
SETNX mykey "v"                    # 仅当不存在时才设置（分布式锁基础）
GETSET mykey "new"                 # 设置新值，返回旧值（已被 SET ... GET 取代）

# 计数（原子操作！）
SET counter 0
INCR counter                       # → 1
INCR counter                       # → 2
INCRBY counter 10                  # → 12
DECR counter                       # → 11
INCRBYFLOAT price 3.14             # 浮点数加减

# 批量
MSET k1 v1 k2 v2 k3 v3
MGET k1 k2 k3                      # 一次拿多个，省 RTT

# 字符串拼接 / 子串
APPEND mykey " world"              # 末尾追加
GETRANGE mykey 0 4                 # 取下标 [0,4]，类似 substring
STRLEN mykey                       # 长度

# 位操作（第 3 章 Bitmap 会详讲）
SETBIT online:20260417 100 1
GETBIT online:20260417 100
BITCOUNT online:20260417

2.1.3 底层编码：int / embstr / raw

┌────────────┬─────────────────────────────────┬────────────────────┐
│  编码       │  使用条件                         │  内存布局           │
├────────────┼─────────────────────────────────┼────────────────────┤
│  int       │  值是整数 + 能用 long 表示         │  直接存 long 值     │
│  embstr    │  字符串 ≤ 44 字节                 │  redisObject 与    │
│            │  （Redis 7 起为 44，老版本是 39） │  SDS 紧挨着分配      │
│  raw       │  字符串 > 44 字节                 │  redisObject + 独立 │
│            │                                 │  分配的 SDS         │
└────────────┴─────────────────────────────────┴────────────────────┘

实测一下：

bash

127.0.0.1:6379> SET k1 100
127.0.0.1:6379> OBJECT ENCODING k1
"int"

127.0.0.1:6379> SET k2 "hello"
127.0.0.1:6379> OBJECT ENCODING k2
"embstr"

127.0.0.1:6379> SET k3 "this is a very long long long long string over 44 bytes!"
127.0.0.1:6379> OBJECT ENCODING k3
"raw"

2.1.4 SDS：Redis 自研的字符串结构

Redis 没有直接用 C 的 char*，而是设计了 SDS（Simple Dynamic String，简单动态字符串）：

C 字符串 (char*)：
  ┌───┬───┬───┬───┬───┬────┐
  │ h │ e │ l │ l │ o │ \0 │
  └───┴───┴───┴───┴───┴────┘
  ❌ 取长度 O(N)（要遍历到 \0）
  ❌ 二进制不安全（遇到 \0 就截断）
  ❌ 拼接易溢出（必须手动 realloc）

Redis SDS（Redis 7 用 sdshdr5/8/16/32/64 五种头部）：
  ┌──────┬──────┬──────┬──────┬──────────────────┬────┐
  │ len  │ alloc│ flags│ ...  │   buf[](实际数据) │ \0 │
  ├──────┼──────┼──────┼──────┼──────────────────┼────┤
  │  5   │  10  │ ...  │      │  hello           │    │
  └──────┴──────┴──────┴──────┴──────────────────┴────┘
                                       ↑
                                  指针指向这里
  ✅ O(1) 取长度（直接读 len）
  ✅ 二进制安全（按 len 读，不看 \0）
  ✅ 预分配 + 惰性释放（减少 realloc）
  ✅ 末尾保留 \0，兼容 C 字符串函数

SDS 扩容策略（关键）：

字符串小于 1 MB：新长度 = 实际需要 × 2（翻倍预分配）
字符串大于等于 1 MB：每次额外多分配 1 MB

这样 N 次 APPEND 的均摊复杂度是 O(N) 而不是 O(N²)。

2.1.5 应用场景

场景	用法
缓存对象	`SET user:1001 "{json}"`
计数器	`INCR article:123:views`
分布式 ID	`INCR global:order:id`（小心单点）
限流	`INCR + EXPIRE` 实现「每秒最多 N 次」
分布式锁	`SET lock NX PX 30000`（详见 Ch11）
Session	`SETEX session:abc 1800 "{json}"`

2.2 List：双端可进可出的有序列表

2.2.1 是什么 + 生活类比

List 是一个有序的字符串列表，两端都可以插入/弹出。

🚇 生活类比：地铁车厢里的一排座位。
LPUSH = 从车头门挤进去
RPUSH = 从车尾门挤进去
LPOP / RPOP = 从相应方向出去
LRANGE = 看哪几个座位上是谁

2.2.2 常用命令

bash

# 入队
LPUSH mylist a b c                 # 头插：list 变为 c b a
RPUSH mylist x y                   # 尾插：list 变为 c b a x y

# 查看
LRANGE mylist 0 -1                 # 取全部：c b a x y
LLEN mylist                        # 长度
LINDEX mylist 0                    # 按下标取：c

# 出队
LPOP mylist                        # 从头弹：c
RPOP mylist                        # 从尾弹：y
LPOP mylist 2                      # 弹 2 个（6.2+）

# 阻塞式弹出（消息队列基础）
BLPOP mylist 5                     # 阻塞 5 秒等数据
BRPOP mylist 0                     # 0 表示永久阻塞

# 修剪 / 删除
LTRIM mylist 0 99                  # 只保留前 100 条（控制队列长度）
LREM mylist 2 a                    # 从头删 2 个值为 a 的元素

2.2.3 底层编码：quicklist 演进史

Redis 3.2 之前：
  ziplist（小数据）  ←──────→  linkedlist（大数据）
  内存紧凑              访问快但内存占用大（每节点 16 字节指针）

Redis 3.2 ~ 6.x：
  quicklist：双向链表 + 每节点内挂一个 ziplist
  既保留链表灵活性，又用 ziplist 压缩内存

Redis 7 之后：
  quicklist 内部的 ziplist 升级为 listpack
  （ziplist 的级联更新问题被 listpack 解决）

quicklist 内部结构：

quicklist：双向链表，每个节点是一个 listpack
┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐
│listpack │ ⇄  │listpack │ ⇄  │listpack │ ⇄  │listpack │
│[a,b,c,d]│    │[e,f,g]  │    │[h,i,j,k]│    │[l,m]    │
└─────────┘    └─────────┘    └─────────┘    └─────────┘

   每个 listpack 默认最多放 128 个元素
   或单 listpack 总字节数 ≤ 8KB（list-max-listpack-size = -2）

为什么这么设计？

纯链表：每节点至少 16 字节指针开销，存大量小元素巨费内存。
纯 ziplist：所有元素放一块连续内存，插入/删除大列表时 realloc 代价高。
quicklist = 两者折中：链表节点数适中（O(N/M)），每节点内紧凑存储。

2.2.4 应用场景

场景	用法
消息队列	`LPUSH` 生产、`BRPOP` 消费（生产者/消费者模型）
最新 N 条数据	`LPUSH` + `LTRIM 0 99` 保留最新 100 条
时间线 / Feed 流	每个用户一个 List，存 post_id
栈 / 队列	同侧 PUSH+POP 是栈，异侧是队列

⚠️ 注意：从 Redis 5.0 起，正式的消息队列推荐用 Stream，List 仅适合简单场景。

2.3 Hash：嵌套的小型 KV

2.3.1 是什么 + 生活类比

Hash 是「KV 里再嵌套一层 KV」，等于把一个对象的字段扁平化存储。

🗄️ 生活类比：一个文件柜（Key）里有很多抽屉（Field），每个抽屉放一份文件（Value）。

String 存对象：
  user:1001 → "{name:'Alice',age:30,city:'BJ'}"
  改一个字段：取出 → 反序列化 → 改 → 序列化 → 存回（5 步）

Hash 存对象：
  user:1001 → { name: "Alice", age: "30", city: "BJ" }
  改一个字段：HSET user:1001 age 31  （1 步，原子）

2.3.2 常用命令

bash

HSET user:1001 name Alice age 30 city BJ
HGET user:1001 name                # → Alice
HMGET user:1001 name age           # 一次取多个字段
HGETALL user:1001                  # 取全部（小心大 Hash！）
HLEN user:1001                     # 字段数
HEXISTS user:1001 email            # 字段是否存在
HDEL user:1001 city                # 删字段
HKEYS user:1001                    # 所有字段名
HVALS user:1001                    # 所有字段值
HINCRBY user:1001 age 1            # 字段值加减（原子）
HSCAN user:1001 0                  # 渐进式遍历（避免 HGETALL 大 Hash）

2.3.3 底层编码：listpack ⇄ hashtable

小 Hash → listpack（默认）
  - 字段数 ≤ 128（hash-max-listpack-entries）
  - 任一字段名/值 ≤ 64 字节（hash-max-listpack-value）
  - 紧凑存储：[field1, value1, field2, value2, ...]
  - O(N) 查找，但 N 小、CPU 缓存友好，比哈希表还快

大 Hash → hashtable
  - 标准的「数组 + 链表」哈希表
  - O(1) 查找
  - 用 dict 结构（Redis 自实现，渐进式 rehash）

编码切换不可逆：一旦升级到 hashtable，即使后续删字段降到 < 128 也不会降回 listpack。

2.3.4 应用场景

场景	用法
存储对象	用户、商品、订单的字段化存储
购物车	`HSET cart:user:1 商品ID 数量`
配置管理	`HSET config:app theme dark lang zh`
频次统计	`HINCRBY stats:2026-04 article:123 1`

2.4 Set：无序去重集合

2.4.1 是什么 + 生活类比

Set 是无序、不重复的字符串集合，类似数学中的集合。

🎟️ 生活类比：演唱会门票存根盒——每张票号唯一，重复投入不会变多，盒子里不分先后顺序。

2.4.2 常用命令

bash

SADD myset a b c                   # 添加（自动去重）
SREM myset a                       # 删除
SMEMBERS myset                     # 所有成员
SCARD myset                        # 元素数
SISMEMBER myset b                  # 是否存在 → 1
SMISMEMBER myset a b c             # 批量查（6.2+）
SRANDMEMBER myset 2                # 随机抽 2 个（不删）
SPOP myset 1                       # 随机弹 1 个（删除）

# 集合运算（强大！）
SINTER set1 set2 set3              # 交集 → 共同好友
SUNION set1 set2                   # 并集
SDIFF  set1 set2                   # 差集 → 我有他没有
SINTERSTORE result set1 set2       # 交集结果存到新 Key

2.4.3 底层编码：intset / listpack / hashtable

全是整数 + 元素 ≤ 512    → intset
                           （有序整数数组，二分查找 O(log N)）
                           
全字符串 + 数量 ≤ 128 + 长度 ≤ 64
                         → listpack（7.0+，老版本是 ziplist）

其他（含字符串且数量大）  → hashtable
                           （用 dict，value 都是 NULL）

intset 内存布局：

struct intset {
  uint32_t encoding;   // INTSET_ENC_INT16 / INT32 / INT64
  uint32_t length;     // 元素数
  int8_t   contents[]; // 按编码大小紧凑排列的有序数组
};

例如 SADD s 1 100 5：
  contents = [1, 5, 100]   ← 始终保持升序
  → SISMEMBER s 100：二分查找
  
当 SADD 一个超过当前编码范围的整数（如 65536）：
  整体「升级」到 INT32，所有元素重新放大
  ↑ 升级不可逆！只升不降

2.4.4 应用场景

场景	用法
标签系统	`SADD tag:user:1 redis mysql golang`
点赞 / 关注	`SADD likes:article:123 user_id` + `SISMEMBER` 判重
共同好友	`SINTER friends:userA friends:userB`
抽奖	`SRANDMEMBER` 不重复抽 / `SPOP` 抽完移除
UV 粗略统计	`SADD page:2026-04-17 user_id`（精确但耗内存，更建议 HyperLogLog）

2.5 ZSet：有序集合（Redis 的「皇冠明珠」）

2.5.1 是什么 + 生活类比

ZSet（Sorted Set）= Set 去重 + 每个元素带一个 score（分数）排序。

🏆 生活类比：游戏排行榜。每个玩家 ID 唯一（去重），按积分（score）从低到高自动排好序。新玩家上分自动插入到合适位置。

2.5.2 常用命令

bash

ZADD rank 100 alice 200 bob 150 charlie    # score member 配对
ZRANGE rank 0 -1 WITHSCORES                # 按分升序：alice/100, charlie/150, bob/200
ZREVRANGE rank 0 2 WITHSCORES              # 按分降序，前 3 名（榜单！）
ZRANGEBYSCORE rank 100 180                 # 分数在 [100,180] 之间的成员
ZRANK rank alice                           # alice 排名（升序）→ 0
ZREVRANK rank alice                        # 降序排名 → 2
ZSCORE rank alice                          # 取分数 → 100
ZINCRBY rank 50 alice                      # alice 加 50 分（原子）→ 150

ZREM rank charlie                          # 移除
ZCARD rank                                 # 元素数
ZCOUNT rank 100 200                        # 分数在区间内的元素数

# 范围操作（Redis 6.2+ 新版 ZRANGE 已能覆盖以下大部分）
ZRANGEBYLEX rank - +                       # 按字典序（score 必须相同时有意义）
ZPOPMIN rank                               # 弹出分数最小（延时队列基础）
ZPOPMAX rank
BZPOPMIN rank 5                            # 阻塞版

2.5.3 底层编码：listpack ⇄ skiplist + hashtable

小 ZSet → listpack
  - 元素数 ≤ 128（zset-max-listpack-entries）
  - 任一 member ≤ 64 字节
  - 紧凑存：[member1, score1, member2, score2, ...]

大 ZSet → skiplist + hashtable（双结构！）
  - skiplist：负责按 score 排序、范围查询 O(log N)
  - hashtable：负责 member → score 的 O(1) 查找
  - 两者数据指针共享，没有重复存储

2.5.4 跳表（Skip List）：为什么不用红黑树？

跳表是 ZSet 的灵魂。来看看它长什么样：

                                              ← Level 4（最稀疏）
头────────────────────────────────────►NULL
   │
   ▼
Level 3:  头──────►[7]───────────────►[42]──►NULL
                    │                  │
                    ▼                  ▼
Level 2:  头──►[3]──►[7]────►[19]────►[42]──►NULL
                │    │        │        │
                ▼    ▼        ▼        ▼
Level 1:  头──►[3]──►[7]──►[12]──►[19]──►[26]──►[42]──►NULL
              ↑       ↑                          ↑
            底层是有序链表                     找 42 的过程：
                                              L4 头→[7]→NULL，回退
                                              L3 [7]→[42]，命中
                                              查询 O(log N)！

跳表 vs 红黑树：

维度	跳表	红黑树
平均复杂度	O(log N)	O(log N)
范围查询	✅ 底层有序链表，直接顺序走	❌ 需要中序遍历，复杂
实现难度	✅ 简单（200 行代码）	❌ 旋转规则复杂
内存	每节点平均 ~1.33 个指针	固定 2~3 个指针
并发优化	✅ 易加锁/无锁化	❌ 旋转影响多个节点

结论：Redis 选跳表是因为范围查询多 + 实现简单，而 ZSet 的核心场景（排行榜按分段查询）正是跳表的强项。

2.5.5 应用场景

场景	用法
排行榜	`ZADD rank score user` + `ZREVRANGE rank 0 9 WITHSCORES`
延时队列	score = 到期时间戳，定期 `ZRANGEBYSCORE -inf 当前时间`
优先级队列	score = 优先级，`ZPOPMIN` 取最高优
滑动窗口限流	每次请求 `ZADD key now now`，`ZREMRANGEBYSCORE` 清过期
附近的人	Geo 类型底层就是 ZSet（详见 Ch3）

2.6 五大类型横向对比速查

┌──────┬─────────┬──────────────────┬──────────────┬──────────────┐
│ 类型 │ 唯一性  │ 顺序             │ 主要复杂度    │ 典型场景      │
├──────┼─────────┼──────────────────┼──────────────┼──────────────┤
│String│ 单值     │ 无               │ O(1)          │ 缓存/计数     │
│List  │ 可重复   │ 插入顺序          │ 头尾O(1)/中间O(N)│队列/时间线   │
│Hash  │ 字段唯一 │ 无               │ O(1)          │ 对象存储      │
│Set   │ 元素唯一 │ 无               │ O(1)          │ 标签/去重     │
│ZSet  │ 元素唯一 │ 按 score 排序     │ O(log N)      │ 排行榜/延时   │
└──────┴─────────┴──────────────────┴──────────────┴──────────────┘

2.7 实操：跑一遍配套代码

实战代码见 02_basic_types/code/：

01_string.py：字符串编码切换实测、INCR 原子性、SDS 预分配观察
02_list_queue.py：用 List 做生产者-消费者消息队列
03_hash_object.py：用 Hash 存对象 vs JSON 字符串的对比
04_set_tags.py：标签系统 + 共同好友
05_zset_leaderboard.py：游戏排行榜实战

浏览器演示见 02_basic_types/demo.html：

① 5 种类型的命令交互沙盒
② Hash 编码 listpack ↔ hashtable 切换可视化
③ 跳表插入查找的动画

2.8 本章小结

┌────────────────────────────────────────────────────────┐
│                     本章核心要点                          │
├────────────────────────────────────────────────────────┤
│                                                          │
│  ① 5 种类型对应 N 种底层编码，小数据用紧凑编码省内存       │
│                                                          │
│  ② SDS 三大优势：O(1) 取长度、二进制安全、预分配          │
│                                                          │
│  ③ List 演进：ziplist+linkedlist → quicklist            │
│                                                          │
│  ④ Hash/Set/ZSet 的小数据编码 7.0 起统一为 listpack     │
│                                                          │
│  ⑤ ZSet 用「跳表 + 哈希表」双结构：                       │
│      - 跳表：O(log N) 范围查询                            │
│      - 哈希表：O(1) 取分数                                │
│                                                          │
│  ⑥ 编码升级不可逆，只升不降                                │
│                                                          │
└────────────────────────────────────────────────────────┘

2.9 面试高频题

Q1：Redis 的 String 最大能存多大？为什么？

考察点：对 String 底层的了解。

标准答案：

单个 String 最大 512 MB（编译期硬编码 512 * 1024 * 1024）。
但实际生产不建议超过 10 KB。原因：
1. 单线程模型下，操作大 Key 会阻塞所有其他请求；
2. 网络传输和序列化耗时；
3. 内存碎片化加剧。

加分项：提到 SDS 的扩容策略（< 1MB 翻倍，>= 1MB 每次加 1MB）。

Q2：String 的 embstr 和 raw 有什么区别？为什么是 44 字节？

考察点：对内存布局的细节理解。

标准答案：

embstr：redisObject 和 SDS 在一次 malloc 中连续分配，CPU 缓存友好，只读。
raw：redisObject 和 SDS 分两次 malloc，可写。
44 字节阈值：jemalloc 的 64 字节内存块去掉 redisObject（16 字节）和 SDS 头部（3 字节）和末尾 \0（1 字节），刚好剩 44 字节给字符串内容。
embstr 修改后会自动转为 raw（因为 embstr 是只读优化）。

Q3：Hash 的底层结构是什么？什么情况下会从 listpack 转为 hashtable？

考察点：编码切换条件。

标准答案：

两种编码：

listpack（7.0+）：紧凑列表，O(N) 查找。
hashtable：标准哈希表，O(1) 查找。

满足以下任一条件会从 listpack 升级为 hashtable：

字段数 > hash-max-listpack-entries（默认 128）
任一字段名或值的长度 > hash-max-listpack-value（默认 64 字节）

注意：升级不可逆，即使后续删字段也不会回到 listpack。

加分项：Redis 7 之前是 ziplist，因为 ziplist 存在「连锁更新」问题（修改某个元素长度可能引发后续所有元素的 cascading update），所以替换为 listpack。

Q4：ZSet 为什么用跳表，不用红黑树？

考察点：数据结构选型理解。

标准答案（Redis 作者 antirez 在 GitHub 上有原话）：

范围查询友好：跳表底层是有序链表，ZRANGEBYSCORE 这类范围操作可以直接顺序遍历，红黑树需要中序遍历，实现复杂。
实现简单：跳表 200 行代码搞定，红黑树需要复杂的旋转和着色规则。
内存可控：跳表平均每节点 1.33 个指针（概率 1/2），红黑树每节点固定 2~3 个指针。
并发友好：跳表的局部修改更易加锁/无锁化。

加分项：补充「ZSet 实际是跳表 + 哈希表的双结构」——跳表负责排序和范围查询，哈希表负责 ZSCORE 这类 O(1) 查找。

Q5：Set 和 ZSet 都能去重，怎么选？

考察点：场景判断。

标准答案：

只需要去重，不关心顺序：用 Set（更省内存，O(1) 查找）。
需要按某个分数排序：用 ZSet（O(log N)，多了一层跳表开销）。
判断元素是否存在：Set 的 SISMEMBER 比 ZSet 的 ZSCORE 略快。

典型场景：

用户标签 → Set
排行榜 → ZSet
已点赞用户 → Set（如果还要按点赞时间排序，则 ZSet 用时间戳做 score）

Q6：用 Redis 实现一个排行榜该怎么做？

考察点：综合应用能力。

标准答案：

python

# 1. 用户加分
ZINCRBY game:rank 10 "alice"

# 2. 取 Top 10
ZREVRANGE game:rank 0 9 WITHSCORES

# 3. 查询某用户排名（降序，第 1 名 rank=0）
ZREVRANK game:rank "alice"

# 4. 查询某用户分数
ZSCORE game:rank "alice"

# 5. 周/月榜：用日期做 Key
ZADD game:rank:2026W16 score user
EXPIRE game:rank:2026W16 86400 * 14   # 自动清理

优化点：

百万级排行榜：分桶（按分数段拆多个 ZSet），降低单 Key 大小。
前 N 名固定缓存：ZREVRANGE 0 99 结果缓存到本地，减少 Redis 压力。
同分排序：score 用 score * 1e7 + (1e7 - timestamp)，让先达到的排前面。

📌 下一章预告：第 3 章我们看 Redis 的「三大特殊数据类型」——Bitmap（1 bit 存海量布尔状态）、HyperLogLog（12 KB 估算 2^64 基数）、Geo（附近的人）。每一个都是「神奇的小工具」。

🎬 可视化演示

演示加载缓慢或样式异常？点此在新标签页打开 ↗

💻 示例代码

python

"""
Ch2 配套代码 1 / 5 —— String 实战

演示：
  1. 三种编码（int / embstr / raw）的自动切换
  2. INCR 的原子性 —— 多线程并发也不会丢更新
  3. SDS 预分配观察 —— 一次分配能撑多次 APPEND
"""

import threading
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)


def section(title: str) -> None:
    print("\n" + "=" * 60)
    print(title)
    print("=" * 60)


def demo_encoding() -> None:
    section("Demo 1: String 三种底层编码")

    cases = {
        "k_int":    "100",
        "k_embstr": "hello",
        "k_raw":    "x" * 50,
    }
    for k, v in cases.items():
        r.set(k, v)
        print(f"{k:10} value={v[:20]!r:25} encoding={r.object('encoding', k)}")

    print("\n  💡 修改 embstr 后会变成 raw（embstr 是只读优化）")
    r.append("k_embstr", " world")
    print(f"  k_embstr after APPEND -> encoding={r.object('encoding', 'k_embstr')}")

    r.delete(*cases.keys())


def demo_incr_atomic(threads: int = 10, per_thread: int = 1000) -> None:
    section(f"Demo 2: INCR 原子性 ({threads} 线程 × {per_thread} 次)")

    key = "demo:counter"
    r.set(key, 0)

    def worker():
        rr = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
        for _ in range(per_thread):
            rr.incr(key)

    ts = [threading.Thread(target=worker) for _ in range(threads)]
    for t in ts: t.start()
    for t in ts: t.join()

    final = int(r.get(key))
    expected = threads * per_thread
    print(f"  expected = {expected}")
    print(f"  actual   = {final}")
    print(f"  ✅ 原子无丢失" if final == expected else "  ❌ 出现丢失（不该发生）")
    r.delete(key)


def demo_sds_prealloc() -> None:
    section("Demo 3: SDS 预分配观察")
    key = "demo:sds"
    r.delete(key)

    sizes = []
    r.set(key, "")
    for i in range(1, 6):
        r.append(key, "x" * 100)
        # MEMORY USAGE 返回的是整个 key 占用的字节数（含开销）
        usage = r.memory_usage(key)
        sizes.append((i * 100, usage))
        print(f"  字符串长度 {i*100:5} 字节  →  实际占用 {usage:5} 字节")

    print("  💡 长度线性增长，但占用是阶梯式跳跃 → SDS 预分配在工作")
    r.delete(key)


if __name__ == "__main__":
    try:
        demo_encoding()
        demo_incr_atomic()
        demo_sds_prealloc()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败: {e}")

python

"""
Ch2 配套代码 2 / 5 —— List 实战：用 BRPOP 实现一个最小生产者-消费者

  - 生产者用 LPUSH 投递任务
  - 消费者用 BRPOP（阻塞式 RPOP）从队列尾部取任务
  - 这是 Stream 出现前最常见的 Redis 消息队列方案

运行：在两个终端分别跑 python 02_list_queue.py producer / consumer
或直接运行本脚本，会模拟单进程内的并发 demo。
"""

import sys
import time
import threading
import redis


QUEUE = "demo:queue"
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)


def producer(n: int = 10) -> None:
    print(f"[producer] 开始投递 {n} 个任务")
    for i in range(n):
        msg = f"task-{i}-{int(time.time())}"
        r.lpush(QUEUE, msg)
        print(f"[producer] LPUSH {msg}  (queue len={r.llen(QUEUE)})")
        time.sleep(0.3)
    print("[producer] 投递完毕")


def consumer(name: str, timeout_sec: int = 5) -> None:
    print(f"[{name}] 开工，BRPOP 等任务（{timeout_sec}s 超时退出）")
    while True:
        item = r.brpop(QUEUE, timeout=timeout_sec)
        if item is None:
            print(f"[{name}] 队列空了，退出")
            return
        _, msg = item
        print(f"[{name}] 处理 {msg}")
        time.sleep(0.5)  # 模拟业务


def demo_concurrent() -> None:
    r.delete(QUEUE)
    threads = [
        threading.Thread(target=producer, kwargs={"n": 8}),
        threading.Thread(target=consumer, args=("consumer-A",)),
        threading.Thread(target=consumer, args=("consumer-B",)),
    ]
    for t in threads: t.start()
    for t in threads: t.join()


if __name__ == "__main__":
    role = sys.argv[1] if len(sys.argv) > 1 else "demo"
    try:
        if role == "producer":
            producer()
        elif role == "consumer":
            consumer("consumer-cli")
        else:
            demo_concurrent()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败: {e}")

python

"""
Ch2 配套代码 3 / 5 —— Hash 实战：对象存储 vs JSON 字符串

对比两种存储方式，体会 Hash 的优势：
  方案 A：String + JSON       —— 改一个字段要全量取/反序列化/序列化/写回
  方案 B：Hash                 —— 字段级原子读写
"""

import json
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 demo_string_json() -> None:
    section("方案 A: String + JSON（修改一个字段要全量序列化）")
    user = {"name": "Alice", "age": 30, "city": "BJ", "vip": True}
    r.set("user:A", json.dumps(user))

    start = time.perf_counter()
    for _ in range(1000):
        # 修改 age：取 → 反序列化 → 改 → 序列化 → 写
        obj = json.loads(r.get("user:A"))
        obj["age"] += 1
        r.set("user:A", json.dumps(obj))
    cost = time.perf_counter() - start
    print(f"  1000 次「age + 1」耗时 {cost*1000:.1f} ms")
    print(f"  最终 age = {json.loads(r.get('user:A'))['age']}")
    r.delete("user:A")


def demo_hash() -> None:
    section("方案 B: Hash（HINCRBY 原子，无序列化）")
    r.hset("user:B", mapping={"name": "Alice", "age": 30, "city": "BJ", "vip": "1"})

    start = time.perf_counter()
    for _ in range(1000):
        r.hincrby("user:B", "age", 1)
    cost = time.perf_counter() - start
    print(f"  1000 次「HINCRBY age 1」耗时 {cost*1000:.1f} ms")
    print(f"  最终 age = {r.hget('user:B', 'age')}")
    print(f"  完整对象 = {r.hgetall('user:B')}")
    print(f"  编码: {r.object('encoding', 'user:B')}")
    r.delete("user:B")


def demo_encoding_switch() -> None:
    section("Hash 编码切换：listpack → hashtable")
    key = "demo:hash"
    r.delete(key)

    # 小 Hash → listpack
    for i in range(5):
        r.hset(key, f"f{i}", f"v{i}")
    print(f"  5 个字段：encoding = {r.object('encoding', key)}")

    # 撑爆字段数（默认 128）
    for i in range(5, 130):
        r.hset(key, f"f{i}", f"v{i}")
    print(f"  130 个字段：encoding = {r.object('encoding', key)}")

    # 即使删回去，编码也不降级
    for i in range(120, 130):
        r.hdel(key, f"f{i}")
    print(f"  删回 120 字段：encoding = {r.object('encoding', key)}  (升级不可逆)")
    r.delete(key)


if __name__ == "__main__":
    try:
        demo_string_json()
        demo_hash()
        demo_encoding_switch()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败: {e}")

python

"""
Ch2 配套代码 4 / 5 —— Set 实战：标签系统 + 共同好友

  - SADD 添加标签 / 关注关系
  - SISMEMBER 判断关注
  - SINTER 算共同好友（集合运算的强项）
"""

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 demo_tags() -> None:
    section("Demo 1: 给文章打标签 + 按标签找文章")

    # 文章 → 标签
    r.sadd("article:1:tags", "redis", "cache", "backend")
    r.sadd("article:2:tags", "mysql", "backend", "sql")
    r.sadd("article:3:tags", "redis", "distributed", "lock")

    # 倒排：标签 → 文章列表
    for art_id in [1, 2, 3]:
        for tag in r.smembers(f"article:{art_id}:tags"):
            r.sadd(f"tag:{tag}:articles", art_id)

    print("  文章 1 的标签：", r.smembers("article:1:tags"))
    print("  redis 标签下的文章：", r.smembers("tag:redis:articles"))
    print("  backend 标签下的文章：", r.smembers("tag:backend:articles"))

    print("\n  💡 同时打 redis + backend 的文章：",
          r.sinter("tag:redis:articles", "tag:backend:articles"))

    for art_id in [1, 2, 3]:
        r.delete(f"article:{art_id}:tags")
    for tag in ["redis", "cache", "backend", "mysql", "sql", "distributed", "lock"]:
        r.delete(f"tag:{tag}:articles")


def demo_common_friends() -> None:
    section("Demo 2: 共同好友 / 可能认识的人")

    r.sadd("friends:alice", "bob", "charlie", "dave", "eve")
    r.sadd("friends:bob",   "alice", "charlie", "frank", "grace")

    print("  Alice 的好友：", r.smembers("friends:alice"))
    print("  Bob   的好友：", r.smembers("friends:bob"))

    print("  共同好友（SINTER）：",
          r.sinter("friends:alice", "friends:bob"))

    print("  Alice 有但 Bob 没有的（SDIFF）→ 可能推荐给 Bob 认识：",
          r.sdiff("friends:alice", "friends:bob") - {"bob"})

    r.delete("friends:alice", "friends:bob")


def demo_intset() -> None:
    section("Demo 3: 全是整数 → intset 编码")
    key = "demo:set"
    r.delete(key)

    r.sadd(key, *range(1, 100))
    print(f"  全是整数：encoding = {r.object('encoding', key)}")

    r.sadd(key, "hello")
    print(f"  混入字符串：encoding = {r.object('encoding', key)}")
    r.delete(key)


if __name__ == "__main__":
    try:
        demo_tags()
        demo_common_friends()
        demo_intset()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败: {e}")

python

"""
Ch2 配套代码 5 / 5 —— ZSet 实战：游戏排行榜

  - ZADD / ZINCRBY 上分
  - ZREVRANGE 取 Top N
  - ZREVRANK 查询某人排名
  - 同分排序的工业级技巧（分数高位放真实分，低位放反向时间戳）
"""

import time
import random
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
KEY = "demo:leaderboard"


def section(t): print("\n" + "=" * 60 + f"\n{t}\n" + "=" * 60)


def setup() -> None:
    r.delete(KEY)
    players = ["alice", "bob", "charlie", "dave", "eve",
               "frank", "grace", "henry", "ivy", "jack"]
    for p in players:
        r.zadd(KEY, {p: random.randint(50, 500)})


def demo_basic() -> None:
    section("Demo 1: 排行榜基础操作")
    setup()

    print("  完整榜单（降序）:")
    for i, (member, score) in enumerate(r.zrevrange(KEY, 0, -1, withscores=True), 1):
        print(f"    #{i:<2} {member:<10} {int(score)}")

    print(f"\n  Top 3: {r.zrevrange(KEY, 0, 2, withscores=True)}")
    print(f"  Alice 排名（降序，0 起）: {r.zrevrank(KEY, 'alice')}")
    print(f"  Alice 当前分数:          {r.zscore(KEY, 'alice')}")
    print(f"  分数在 [200, 400] 的人:  {r.zrangebyscore(KEY, 200, 400)}")


def demo_incr() -> None:
    section("Demo 2: ZINCRBY 上分（原子）")
    before = r.zscore(KEY, "alice")
    r.zincrby(KEY, 50, "alice")
    after = r.zscore(KEY, "alice")
    print(f"  Alice {before} → {after}（加 50）")
    print(f"  最新排名: #{r.zrevrank(KEY, 'alice') + 1}")


def demo_same_score_trick() -> None:
    section("Demo 3: 同分排序技巧（先到者排前）")
    key = "demo:rank2"
    r.delete(key)

    # 把分数编码为：真实分 * 1e10 + (1e10 - 时间戳)
    # 这样同分时先到者反向时间戳更大，排名更靠前
    def encode_score(score: int, ts: float) -> float:
        return score * 1e10 + (1e10 - int(ts * 1000))

    base_ts = time.time()
    r.zadd(key, {"alice":   encode_score(100, base_ts)})
    time.sleep(0.01)
    r.zadd(key, {"bob":     encode_score(100, time.time())})
    time.sleep(0.01)
    r.zadd(key, {"charlie": encode_score(100, time.time())})

    print("  三人同 100 分，按到达时间排（alice 最早）：")
    for i, m in enumerate(r.zrevrange(key, 0, -1), 1):
        print(f"    #{i} {m}")
    r.delete(key)


def demo_encoding() -> None:
    section("Demo 4: ZSet 编码切换")
    key = "demo:enc_zset"
    r.delete(key)
    r.zadd(key, {f"m{i}": i for i in range(5)})
    print(f"  5 元素：encoding = {r.object('encoding', key)}")

    r.zadd(key, {f"m{i}": i for i in range(5, 130)})
    print(f"  130 元素：encoding = {r.object('encoding', key)}")
    r.delete(key)


if __name__ == "__main__":
    try:
        demo_basic()
        demo_incr()
        demo_same_score_trick()
        demo_encoding()
        r.delete(KEY)
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败: {e}")

01_string.py ↗ · 02_list_queue.py ↗ · 03_hash_object.py ↗ · 04_set_tags.py ↗ · 05_zset_leaderboard.py ↗

第 2 章 五大基础数据类型 ​

2.0 概览：5 种类型 × N 种编码 ​

2.1 String：最简单也最复杂 ​

2.1.1 是什么 + 生活类比 ​

2.1.2 常用命令 ​

2.1.3 底层编码：int / embstr / raw ​

2.1.4 SDS：Redis 自研的字符串结构 ​

2.1.5 应用场景 ​

2.2 List：双端可进可出的有序列表 ​

2.2.1 是什么 + 生活类比 ​

2.2.2 常用命令 ​

2.2.3 底层编码：quicklist 演进史 ​

2.2.4 应用场景 ​

2.3 Hash：嵌套的小型 KV ​

2.3.1 是什么 + 生活类比 ​

2.3.2 常用命令 ​

2.3.3 底层编码：listpack ⇄ hashtable ​

2.3.4 应用场景 ​

2.4 Set：无序去重集合 ​

2.4.1 是什么 + 生活类比 ​

2.4.2 常用命令 ​

2.4.3 底层编码：intset / listpack / hashtable ​

2.4.4 应用场景 ​

2.5 ZSet：有序集合（Redis 的「皇冠明珠」） ​

2.5.1 是什么 + 生活类比 ​

2.5.2 常用命令 ​

2.5.3 底层编码：listpack ⇄ skiplist + hashtable ​

2.5.4 跳表（Skip List）：为什么不用红黑树？ ​

2.5.5 应用场景 ​

2.6 五大类型横向对比速查 ​

2.7 实操：跑一遍配套代码 ​

2.8 本章小结 ​

2.9 面试高频题 ​

Q1：Redis 的 String 最大能存多大？为什么？ ​

Q2：String 的 embstr 和 raw 有什么区别？为什么是 44 字节？ ​

Q3：Hash 的底层结构是什么？什么情况下会从 listpack 转为 hashtable？ ​

Q4：ZSet 为什么用跳表，不用红黑树？ ​

Q5：Set 和 ZSet 都能去重，怎么选？ ​

Q6：用 Redis 实现一个排行榜该怎么做？ ​

🎬 可视化演示 ​

💻 示例代码 ​

第 2 章五大基础数据类型

2.0 概览：5 种类型 × N 种编码

2.1 String：最简单也最复杂

2.1.1 是什么 + 生活类比

2.1.2 常用命令

2.1.3 底层编码：int / embstr / raw

2.1.4 SDS：Redis 自研的字符串结构

2.1.5 应用场景

2.2 List：双端可进可出的有序列表

2.2.1 是什么 + 生活类比

2.2.2 常用命令

2.2.3 底层编码：quicklist 演进史

2.2.4 应用场景

2.3 Hash：嵌套的小型 KV

2.3.1 是什么 + 生活类比

2.3.2 常用命令

2.3.3 底层编码：listpack ⇄ hashtable

2.3.4 应用场景

2.4 Set：无序去重集合

2.4.1 是什么 + 生活类比

2.4.2 常用命令

2.4.3 底层编码：intset / listpack / hashtable

2.4.4 应用场景

2.5 ZSet：有序集合（Redis 的「皇冠明珠」）

2.5.1 是什么 + 生活类比

2.5.2 常用命令

2.5.3 底层编码：listpack ⇄ skiplist + hashtable

2.5.4 跳表（Skip List）：为什么不用红黑树？

2.5.5 应用场景

2.6 五大类型横向对比速查

2.7 实操：跑一遍配套代码

2.8 本章小结

2.9 面试高频题

Q1：Redis 的 String 最大能存多大？为什么？

Q2：String 的 embstr 和 raw 有什么区别？为什么是 44 字节？

Q3：Hash 的底层结构是什么？什么情况下会从 listpack 转为 hashtable？

Q4：ZSet 为什么用跳表，不用红黑树？

Q5：Set 和 ZSet 都能去重，怎么选？

Q6：用 Redis 实现一个排行榜该怎么做？

🎬 可视化演示

💻 示例代码