别急着换模型：上下文工程才是Agent效果的天花板
2026/10/31 8:29:24
Redis 7 天学习笔记 · Day 2
主题：五种数据结构——每种只学“什么时候非它不可”
环境：Ubuntu 虚拟机 + Python 3 + Redis
一句话总结：数据结构选错，代码会变丑；选对，一行命令解决的事不用写十行。
一、今日核心认知
Day 1 只用了 String，把整个用户对象json.dumps塞进去。要改一个email字段，你得读出整个 JSON → 反序列化 → 改字段 → 再序列化 → 写回去。
今天认识四种结构，目标只有一个：拿到一个业务场景，能在 10 秒内判断该用哪个，并且能说清楚“为什么不是另一个”。
先记住三条铁律：
String 是万能，但“万能”往往意味着“不是最优”。结构化数据优先 Hash，去重优先 Set，排序优先 ZSet。
每种结构都有它“会出事”的边界。Hash 字段过多、List 无限增长、Set 成员过多，都会变成 BigKey。
Python 返回值类型不同，
hgetall返回 dict，smembers返回 set，zrange返回 list。
二、五种数据结构速记与 Python 实操
1. String：最基础的万能型，但别滥用
结构特点：一个 key 对应一个 value。value 可以是字符串、数字、二进制。数字支持原子增减。
一句话定位：计数、缓存单个值、验证码。结构化数据不要用它。
import redis r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True) # ===== String 核心操作 ===== # 存一个键值对。set 会覆盖旧值。 r.set("article:42:title", "Redis入门") # 取一个键的值。key 不存在时返回 None，不是报错。 title = r.get("article:42:title") print(title) # Redis入门 # setex = set + expire，存的同时设过期时间（秒）。 # 用于验证码、Token 这类“过一段时间就该消失”的数据。 r.setex("verify:code:13800138000", 300, "123456") # 5分钟后自动消失 # setnx = set if not exists，key 不存在才写入。 # 返回 True 表示写入成功，False 表示 key 已存在、本次没写。 # 分布式锁的雏形就是这个命令。 locked = r.setnx("lock:order:1001", "1") print(locked) # True（第一次） locked2 = r.setnx("lock:order:1001", "1") print(locked2) # False（已存在，没写进去） # incr 是原子加 1。多个请求同时 incr，不会丢计数。 # 用于阅读量、点赞数、库存扣减。 r.set("article:42:views", 100) new_views = r.incr("article:42:views") # 返回 101 print(new_views) # mset / mget 批量操作，减少网络往返次数。 # 一次 mget 拿多个 key，比循环 get 快一个数量级。 r.mset({"user:1:name": "张三", "user:1:age": "25"}) values = r.mget(["user:1:name", "user:1:age"]) print(values) # ['张三', '25']什么时候不该用：要存一个有多个字段的对象（用户、商品），用Hash。每次改一个字段都要读出整个 JSON，既浪费网络又容易并发覆盖。
2. Hash：存对象，改一个字段不用动全身
结构特点：一个 key 下面是一张 field-value 映射表，相当于“key 嵌套了一个字典”。Redis 7 中小的 Hash 用 ListPack 紧凑存储，省内存；字段多了会转成 HashTable。
一句话定位：用户信息、商品属性、配置中心。结构化对象的首选。
# ===== Hash 核心操作 ===== # hset 设置一个字段。 # 返回 1 表示新建了字段，0 表示字段已存在、只是更新了值。 r.hset("user:1001", "name", "张三") # hset 也可以批量设置多个字段，用 mapping 参数。 r.hset("user:1001", mapping={ "age": "25", "email": "zhangsan@example.com", "city": "北京" }) # hget 只取一个字段。 # 改 email 只需要这一次 hset，不用读整个对象再写回去。 r.hset("user:1001", "email", "new@example.com") print(r.hget("user:1001", "email")) # new@example.com # hgetall 拿全部字段，返回的是 Python dict。 # ⚠️ 字段数过万时不要用，会阻塞 Redis。大 Hash 用 hscan 分批遍历。 user = r.hgetall("user:1001") print(user) # {'name': '张三', 'age': '25', 'email': 'new@example.com', 'city': '北京'} # hincrby 对某个字段做原子增减。 # 用于“用户积分 +10”“库存 -1”这类场景。 r.hincrby("user:1001", "age", 1) # age 从 25 变成 26 # hlen 统计字段数量。 print(r.hlen("user:1001")) # 4 # hdel 删除一个或多个字段。 r.hdel("user:1001", "city") # hscan 分批遍历大 Hash，避免 hgetall 一次性拉全部。 # cursor 从 0 开始，返回 0 表示遍历结束。 cursor = 0 while True: cursor, data = r.hscan("user:1001", cursor=cursor, count=10) print(data) if cursor == 0: break坑点：hgetall在字段数很大时会阻塞 Redis。日常用户对象只有几十个字段没问题，但如果一个 Hash 有上万个字段，就必须用 hscan。
3. List：有序可重复，轻量队列
结构特点：双向链表，两端增删都是 O(1)。可以当栈（LIFO）也可以当队列（FIFO）。
一句话定位：最新 N 条记录、简单消息队列、时间线。不要用它做“按索引随机查中间元素”的事，lindex是 O(n)。
# ===== List 核心操作 ===== # rpush 从右边插入。元素顺序：a -> b -> c r.rpush("article:42:comments", "评论A", "评论B", "评论C") # lpush 从左边插入。lpush "D" 后，列表变成 D, A, B, C r.lpush("article:42:comments", "评论D") # lrange 按索引范围取。 # 0 是第一个，-1 是最后一个。lrange 0 -1 取全部。 comments = r.lrange("article:42:comments", 0, -1) print(comments) # ['评论D', '评论A', '评论B', '评论C'] # 取最新 3 条：从左边数前 3 个。 latest = r.lrange("article:42:comments", 0, 2) print(latest) # ['评论D', '评论A', '评论B'] # lpop / rpop 从两端弹出并删除。 # r.lpop 取走最左边；r.rpop 取走最右边。 r.lpop("article:42:comments") # 取走 '评论D' r.rpop("article:42:comments") # 取走 '评论C' # llen 获取列表长度。 print(r.llen("article:42:comments")) # ⚠️ lindex 是 O(n)。不要用 lindex 查一个大 List 的中间元素。 # 如果业务需要“按位置随机访问”，说明 List 选错了。坑点：List 没有“长度上限”。如果只做lpush不做ltrim，这个 List 会无限增长，变成 BigKey。用 List 存“最新 N 条”时，一定配一个ltrim把范围外的裁掉：
# 只保留最新的 100 条，超出范围的自动裁掉。 r.ltrim("article:42:comments", 0, 99)4. Set：无序去重，专治“不重复”和“交集”
结构特点：无序、元素唯一。支持交集、并集、差集。整数集合用 IntSet 紧凑存储，元素多了或出现非整数后转 HashTable。
一句话定位：去重、共同好友、标签、抽奖。不需要排序时，Set 比 ZSet 省内存。
# ===== Set 核心操作 ===== # sadd 添加元素。自动去重，重复的会被忽略。 # 返回实际新加入的元素个数。 r.sadd("user:1001:follows", "1002", "1003", "1004") r.sadd("user:1002:follows", "1001", "1003", "1005") # 重复添加 "1002"，返回 0，集合不变。 r.sadd("user:1001:follows", "1002") # 返回 0 # smembers 查看全部元素，返回 Python set。 print(r.smembers("user:1001:follows")) # {'1002', '1003', '1004'} # sismember 判断元素是否存在，返回 True / False。 # O(1)，比用 List 遍历判断快得多。 print(r.sismember("user:1001:follows", "1002")) # True # sinter 交集：两个用户共同关注了谁。 common = r.sinter("user:1001:follows", "user:1002:follows") print(common) # {'1003'} # sunion 并集：合并去重。 union = r.sunion("user:1001:follows", "user:1002:follows") # sdiff 差集：A 有但 B 没有的。 only_1001 = r.sdiff("user:1001:follows", "user:1002:follows") print(only_1001) # {'1002', '1004'} # srandmember 随机取 N 个，不删除。抽奖“抽中但还能再抽”用这个。 winners = r.srandmember("lottery", 2) # spop 随机弹出 1 个并删除。抽奖“抽走就没了”用这个。 winner = r.spop("lottery")坑点：smembers在大 Set 上同样会阻塞，大 Set 用sscan分批遍历。
5. ZSet：带分数，专为排序而生
结构特点：元素唯一，每个元素带一个 score，按 score 自动排序。底层是跳跃表 + 哈希表，范围查询和排名都是 O(log N)。
一句话定位：排行榜、积分、延时队列。只要业务里有“排名”或“按时间排序”，就是它。
# ===== ZSet 核心操作 ===== # zadd 添加元素，score 在前，值在后。用 dict 一次传多个。 r.zadd("rank:game:2026", { "张三": 95, "李四": 87, "王五": 95, # score 可以重复，但元素名不能重复 "赵六": 72 }) # zrevrange 按 score 从高到低取。 # 0 到 2 取前三名。返回的是元素名列表。 top3 = r.zrevrange("rank:game:2026", 0, 2) print(top3) # ['张三', '王五', '李四']（score 相同时按字典序） # 带分数一起取：加上 withscores=True，返回的是“元素-分数”列表。 top3_with_score = r.zrevrange("rank:game:2026", 0, 2, withscores=True) print(top3_with_score) # [('张三', 95.0), ('王五', 95.0), ('李四', 87.0)] # zscore 查单个元素的分数。 print(r.zscore("rank:game:2026", "张三")) # 95.0 # zrevrank 查排名（从高到低，0 表示第一名）。 print(r.zrevrank("rank:game:2026", "张三")) # 0 # zincrby 对某个元素的 score 原子增加。 # 用于“用户又得了 5 分”“视频播放量 +1”且需要排序的场景。 r.zincrby("rank:game:2026", 5, "李四") # 李四 87 -> 92 # zrangebyscore 按分数范围取。 # 取 90 分以上的所有人（min 和 max 是分数，不是排名）。 high_scorers = r.zrangebyscore("rank:game:2026", 90, 100) print(high_scorers) # ['张三', '王五', '李四'] # zcard 统计元素个数。 print(r.zcard("rank:game:2026")) # 4 # zrem 删除指定元素。 r.zrem("rank:game:2026", "赵六")ZSet 做延时队列的思路：
import time # score 存“任务应该执行的时间戳”。 # 消费端反复查 score ≤ 当前时间的任务，取到就执行，执行完删掉。 now = int(time.time()) r.zadd("delay:queue", {"task:1001": now + 10}) # 10 秒后执行 ready = r.zrangebyscore("delay:queue", 0, int(time.time())) if ready: task = ready[0] # 执行任务... r.zrem("delay:queue", task) # 执行完删除坑点：score 是浮点数。如果业务需要精确的整数排序（比如订单号），最好直接用整数作为 score，避免浮点精度带来的意外并列。
三、场景选型对照表
| 场景 | 选谁 | 为什么不选另一个 |
|---|---|---|
| 文章阅读量 +1、验证码 | String | Hash 做计数器是杀鸡用牛刀；Set 不能存数字值 |
| 存一个用户对象，只改 email | Hash | String 存 JSON 要读出整个改完再写回；改一个字段发一次 hset 就够了 |
| 最新 20 条评论 | List | Set 无序；ZSet 带分数但这里不需要排序 |
| 两个人的共同关注 | Set | List 不能做交集且允许重复；ZSet 的 score 在这里没用 |
| 游戏排行榜前 10 | ZSet | Set 无序，拿不到“前 10”；List 无法按分数排序 |
| 去重（如抽奖名单） | Set | List 允许重复；Hash 的 field 虽然唯一但访问方式不对 |
四、验证实验
代码跑完后，在redis-cli里逐个确认：
TYPE user:1001 # hash TYPE rank:game:2026 # zset TYPE article:42:comments # list TYPE user:1001:follows # set HGETALL user:1001 ZREVRANGE rank:game:2026 0 -1 WITHSCORES LRANGE article:42:comments 0 -1 SMEMBERS user:1001:follows
再用OBJECT ENCODING看一眼底层编码的变化：
OBJECT ENCODING user:1001 # listpack（字段少时）或 hashtable OBJECT ENCODING rank:game:2026 # listpack（元素少时）或 skiplist
五、今日踩坑 / 待确认
| 问题 | 原因 | 处理 |
|---|---|---|
hgetall返回的是 dict，我当列表用了 | redis-py对 Hash 的返回值做了类型映射 | hgetall→ dict；hmget→ list；hget→ str/None |
smembers返回set，不能直接 JSON 序列化 | 返回值是 Pythonset | 先list(r.smembers(...))再序列化 |
ZSet 的 score 取出来是95.0，不是95 | score 本质是浮点数 | 需要整数时自己int()转换 |
hgetall把 Redis 卡住了 | 大 Hash 一次性拉全量字段 | 改用hscan分批遍历 |
| List 越来越大，内存持续上涨 | 只lpush没有ltrim | 用ltrim限制保留范围 |
六、交付物 Checklist
- 截图 1：5 个结构各一段运行输出，能看出
hgetall返回 dict、smembers返回 set、zrevrange withscores返回列表 - 一句话：从对照表里挑一行，解释“为什么不是另一个”