Redis 数据结构与缓存
从”为什么需要它”开始讲，配合通俗比喻和完整可运行命令。所有 Redis 命令都可以在
redis-cli中直接敲。
本文定位：本篇是 Redis 五种数据结构（理论篇） 的实战落地 companion——聚焦五种结构的命令实操、缓存三大坑（穿透/击穿/雪崩）的代码防护与常见踩坑；概念原理以理论篇为准。
目录
Redis 是什么
一个场景理解缓存
你在电商网站搜”笔记本电脑”，页面要显示商品列表。如果每次搜索都去查 MySQL：
flowchart TD User["用户搜'笔记本电脑'"] User -->|无缓存| MySQL["查 MySQL<br/>10-50ms"] User -->|有缓存| Redis["查 Redis<br/><1ms"] Redis -->|命中| Return["直接返回"] Redis -->|未命中| MySQL
MySQL 是仓库（空间大、拿东西慢），Redis 是工作台上的工具箱（空间小、伸手就拿）。常用的螺丝刀放工具箱，偶尔用的大电钻才去仓库取。
Redis 为什么快？
| 原因 | 解释 |
|---|---|
| 纯内存操作 | 数据在 RAM 里，不碰硬盘（MySQL 要走磁盘 I/O） |
| 单线程模型 | 没有锁竞争、没有上下文切换开销 |
| IO 多路复用 | 一个线程同时监听多个连接，哪个有数据到了就处理哪个 |
| 简单数据结构 | 没有复杂的关系模型，就是 key-value |
单线程为什么还这么快？
Redis 的瓶颈从来不在 CPU，而在网络带宽和内存大小。就像一个收银员速度够快，瓶颈是顾客排多长队、收银台多大——不是收银员脑子转不过来。
QPS 量级概念：MySQL 单机 QPS ≈ 几千到一万，Redis 单机 QPS ≈ 10 万+。不是快一两倍，是快两个数量级。
Redis 和 MySQL 的关系：不是替代，是互补
flowchart TD subgraph 应用服务器 Read[读请求] --> Redis["Redis（内存，快）"] Redis -->|没命中| MySQL["MySQL（硬盘，持久）"] Write[写请求] --> MySQL end
Note
MySQL 是日记本（要长期保存的），Redis 是便利贴（用完可以丢、但随时能看）。
安装运行
Redis 官方不直接支持 Windows。以下三种方式任选一种。
方式 1：Docker（推荐，一行搞定）
# 拉取并运行 Redis（端口 6379 是 Redis 默认端口）
docker run -d --name myredis -p 6379:6379 redis
# 进入 Redis 命令行
docker exec -it myredis redis-cli
方式 2：WSL（Windows Subsystem for Linux）
# 在 WSL Ubuntu 里
sudo apt update
sudo apt install redis-server
# 启动
sudo service redis-server start
# 连接
redis-cli
方式 3：Memurai（Windows 原生，兼容 Redis 协议）
下载地址：https://www.memurai.com/ → 安装后自动启动，直接 redis-cli 连接。
验证安装成功
$ redis-cli
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> SET hello world
OK
127.0.0.1:6379> GET hello
"world"
前置知识：命令行基础
Redis CLI 基本规则
| 规则 | 说明 | 示例 |
|---|---|---|
| 命令不区分大小写 | GET = get（习惯大写） | GET name |
| key 区分大小写 | user ≠ User | |
| 字符串值带引号时含空格 | SET title "hello world" | 双引号保护空格 |
| 默认操作 0 号数据库 | Redis 有 0~15 共 16 个库 | SELECT 1 切换 |
Ctrl + C 退出 | 或直接关终端 |
通用命令（所有数据类型都适用）
| 命令 | 作用 | 示例 |
|---|---|---|
KEYS pattern | 按模式查找 key | KEYS user:* |
DEL key | 删除 key | DEL user:1 |
EXISTS key | 判断 key 是否存在 | EXISTS user:1 |
EXPIRE key seconds | 设置过期时间（秒） | EXPIRE code 300 |
TTL key | 查看剩余过期时间 | TTL code（-1=永不过期，-2=已过期） |
TYPE key | 查看 key 的数据类型 | TYPE user:1 |
RENAME old new | 重命名 key | RENAME tmp name |
FLUSHDB | ⚠️ 清空当前库 | 生产环境禁用！ |
一、String — 字符串
1.1 为什么需要 String？
Redis 最基础的类型。存一个 key 对应一个值：
SET name "张三"
GET name # "张三"
String 不仅能存字符串，还能存数字并自增——这是普通缓存做不到的。
String 就像便利贴——贴一个标签（key），上面写一行字（value）。
1.2 核心命令
| 命令 | 作用 | 示例 |
|---|---|---|
SET key value | 设置值（有则覆盖） | SET name "张三" |
GET key | 取值 | GET name |
MSET k1 v1 k2 v2 | 批量设值 | MSET user:1:name "张三" user:1:age 25 |
MGET k1 k2 | 批量取值 | MGET user:1:name user:1:age |
SETEX key seconds value | 设值 + 设过期时间 | SETEX code 300 "123456" |
SETNX key value | 仅 key 不存在时才设（SET if Not eXists） | SETNX lock:task "1" |
INCR key | 值 +1（值必须是数字） | INCR views |
DECR key | 值 -1 | DECR stock |
INCRBY key N | 值 +N | INCRBY views 10 |
APPEND key value | 追加字符串 | APPEND log "new line" |
STRLEN key | 字符串长度 | STRLEN name |
GETSET key value | 取旧值的同时设新值 | GETSET name "李四" |
# 基本存取
127.0.0.1:6379> SET username "zhangsan"
OK
127.0.0.1:6379> GET username
"zhangsan"
# 计数器（INCR 天生线程安全——没有"读-改-写"的并发问题）
127.0.0.1:6379> SET page:views 0
OK
127.0.0.1:6379> INCR page:views # → 1
(integer) 1
127.0.0.1:6379> INCR page:views # → 2
(integer) 2
127.0.0.1:6379> INCRBY page:views 10 # → 12
(integer) 12
# 验证码场景：存 5 分钟自动过期
127.0.0.1:6379> SETEX sms:13800138000 300 "456789"
OK
127.0.0.1:6379> TTL sms:13800138000 # 还剩多少秒
(integer) 295
# 分布式锁原理：SETNX 只有一个客户端能 set 成功
127.0.0.1:6379> SETNX order:1001 "locked"
(integer) 1 # ← 拿到锁了
# 另一个客户端同时执行：
127.0.0.1:6379> SETNX order:1001 "locked"
(integer) 0 # ← 没拿到，别人锁着呢
1.3 常见场景
| 场景 | 用到的命令 | 说明 |
|---|---|---|
| 缓存 JSON 数据 | SET/GET | 把对象序列化成 JSON 字符串存 |
| 计数器（阅读量/点赞） | INCR | 原子操作，并发安全 |
| 验证码/Token | SETEX | 带过期时间，自动清理 |
| 分布式锁 | SETNX | ”只有第一个执行的能 set 成功” |
| 限流计数器 | INCR + EXPIRE | 统计 1 分钟内请求次数 |
二、Hash — 哈希
2.1 为什么需要 Hash？
String 存对象要序列化成 JSON，想改一个字段就得整体取出→修改→写回。Hash 直接存字段，可以按字段读写。
# ❌ String 存用户：改邮箱要整取整写
SET user:1 '{"name":"张三","age":25,"email":"zs@qq.com"}'
# ✅ Hash 存用户：改邮箱只动一个字段
HSET user:1 name "张三" age 25 email "zs@qq.com"
HSET user:1 email "new@qq.com" # 只改 email
String 存对象就像把整张表格拍成照片存——改一格得重拍。Hash 就像 Excel 表格——直接定位到单元格修改。
2.2 核心命令
| 命令 | 作用 | 示例 |
|---|---|---|
HSET key field value | 设单个字段 | HSET user:1 name "张三" |
HSET key f1 v1 f2 v2 | 设多个字段（Redis 4.0+） | HSET user:1 name "张三" age 25 |
HGET key field | 取单个字段 | HGET user:1 name |
HMGET key f1 f2 | 取多个字段 | HMGET user:1 name age |
HGETALL key | 取所有字段和值 | HGETALL user:1 |
HDEL key field | 删除字段 | HDEL user:1 email |
HEXISTS key field | 判断字段是否存在 | HEXISTS user:1 name |
HKEYS key | 取所有字段名 | HKEYS user:1 |
HVALS key | 取所有字段值 | HVALS user:1 |
HLEN key | 字段数量 | HLEN user:1 |
HINCRBY key field N | 字段值 +N | HINCRBY user:1 age 1 |
# 存用户信息
127.0.0.1:6379> HSET user:1001 name "张三" age 25 email "zs@qq.com"
(integer) 3
# 取单个字段
127.0.0.1:6379> HGET user:1001 name
"张三"
# 取多个字段
127.0.0.1:6379> HMGET user:1001 name age
1) "张三"
2) "25"
# 取全部（field 和 value 交替返回）
127.0.0.1:6379> HGETALL user:1001
1) "name"
2) "张三"
3) "age"
4) "25"
5) "email"
6) "zs@qq.com"
# 判断字段是否存在
127.0.0.1:6379> HEXISTS user:1001 phone
(integer) 0
# 年龄 +1
127.0.0.1:6379> HINCRBY user:1001 age 1
(integer) 26
# 购物车场景：商品 ID → 数量
127.0.0.1:6379> HSET cart:1001 sku_101 2 sku_202 1
(integer) 2
127.0.0.1:6379> HINCRBY cart:1001 sku_101 1
(integer) 3
2.3 常见场景
| 场景 | 用到的命令 | 说明 |
|---|---|---|
| 用户信息缓存 | HSET/HGET/HMGET | 按字段读写，不用整取整写 |
| 购物车 | HSET/HINCRBY/HDEL | 商品 ID 做 field，数量做 value |
| 文章点赞数统计 | HINCRBY | 原子自增，并发安全 |
| 对象属性快速查询 | HEXISTS | 判断用户是否有某属性 |
三、List — 列表
3.1 为什么需要 List？
List 是有序可重复的字符串列表。关键特性：可以从两端 push/pop——天然适合做队列和栈。
List 像一根竹筒——只能从两端塞东西或取东西。从左边塞从右边取 → 队列（先进先出）；从左边塞从左边取 → 栈（后进先出）。
3.2 核心命令
| 命令 | 作用 | 示例 |
|---|---|---|
LPUSH key value | 从左边插入 | LPUSH tasks "send_email" |
RPUSH key value | 从右边插入 | RPUSH tasks "send_sms" |
LPOP key | 从左边弹出（取出并删除） | LPOP tasks |
RPOP key | 从右边弹出 | RPOP tasks |
LRANGE key start stop | 按范围取（不删除） | LRANGE tasks 0 -1（取全部） |
LLEN key | 列表长度 | LLEN tasks |
LINDEX key index | 按下标取 | LINDEX tasks 0（第一个） |
LREM key count value | 删除指定值的元素 | LREM tasks 1 "send_email" |
LTRIM key start stop | 只保留范围内的 | LTRIM news 0 9 |
# 从左边塞入（最新的在最左边）
127.0.0.1:6379> LPUSH news "消息1"
(integer) 1
127.0.0.1:6379> LPUSH news "消息2"
(integer) 2
127.0.0.1:6379> LPUSH news "消息3"
(integer) 3
# 取全部
127.0.0.1:6379> LRANGE news 0 -1
1) "消息3"
2) "消息2"
3) "消息1"
# ===== 队列模式：LPUSH + RPOP（先进先出） =====
127.0.0.1:6379> LPUSH queue "task1" "task2" "task3"
(integer) 3
127.0.0.1:6379> RPOP queue # 右边弹出 = 最早进去的
"task1"
127.0.0.1:6379> RPOP queue
"task2"
# ===== 栈模式：LPUSH + LPOP（后进先出） =====
127.0.0.1:6379> LPUSH stack "a" "b" "c"
(integer) 3
127.0.0.1:6379> LPOP stack # 左边弹出 = 最后进去的
"c"
127.0.0.1:6379> LPOP stack
"b"
# ===== 阻塞弹出：队列空了会等着，有新消息立即返回 =====
127.0.0.1:6379> BLPOP queue 5 # 5秒内没数据就返回 nil
3.3 常见场景
| 场景 | 用到的命令 | 说明 |
|---|---|---|
| 消息队列 | LPUSH + RPOP/BLPOP | 生产者左边塞，消费者右边取 |
| 最新动态/时间线 | LPUSH + LTRIM | 左边塞新的，LTRIM 只保留最近 N 条 |
| 任务栈 | LPUSH + LPOP | 后进先出，深度优先处理 |
| 分页列表 | LRANGE | LRANGE key 0 9 取前 10 条 |
List 不是正经消息队列
没有确认机制，消费者崩了消息就丢了。生产环境消息队列用 RabbitMQ/Kafka。List 适合轻量场景或学习。
四、Set — 集合
4.1 为什么需要 Set？
Set 是无序不重复的字符串集合。核心价值在集合运算——交集、并集、差集，一句命令搞定。
Set 像微信群——群里每个人只能出现一次（不重复），看谁和谁在同一个群就是交集。
4.2 核心命令
| 命令 | 作用 | 示例 |
|---|---|---|
SADD key member | 添加元素 | SADD tags:article:1 "python" |
SREM key member | 删除元素 | SREM tags:article:1 "python" |
SMEMBERS key | 取所有元素 | SMEMBERS tags:article:1 |
SISMEMBER key member | 判断是否存在 | SISMEMBER tags:article:1 "python" |
SCARD key | 元素个数（Set CARDinality） | SCARD tags:article:1 |
SINTER key1 key2 | 交集 — 两个集合都有的 | SINTER user:1:friends user:2:friends |
SUNION key1 key2 | 并集 — 合并去重 | SUNION tags:1 tags:2 |
SDIFF key1 key2 | 差集 — key1 有 key2 没有的 | SDIFF user:1:friends user:2:friends |
SRANDMEMBER key N | 随机取 N 个（不删除） | SRANDMEMBER users 3 |
SPOP key | 随机弹出一个（取出并删除） | SPOP lottery |
# 给文章打标签
127.0.0.1:6379> SADD article:1:tags "python" "redis" "database"
(integer) 3
127.0.0.1:6379> SADD article:2:tags "python" "django" "web"
(integer) 3
# 查看文章 1 的所有标签
127.0.0.1:6379> SMEMBERS article:1:tags
1) "python"
2) "redis"
3) "database"
# 两篇文章的共同标签（交集）
127.0.0.1:6379> SINTER article:1:tags article:2:tags
1) "python"
# ===== 共同好友（交集） =====
127.0.0.1:6379> SADD user:1:friends "user:2" "user:3" "user:5"
(integer) 3
127.0.0.1:6379> SADD user:4:friends "user:2" "user:5" "user:8"
(integer) 3
127.0.0.1:6379> SINTER user:1:friends user:4:friends
1) "user:2"
2) "user:5"
# user:4 的好友里，user:1 还没有的是谁？
127.0.0.1:6379> SDIFF user:4:friends user:1:friends
1) "user:8" # user:1 不认识 user:8 → 推荐
# ===== 抽奖（随机弹出） =====
127.0.0.1:6379> SADD lottery "user:1" "user:2" "user:3" "user:4" "user:5"
(integer) 5
127.0.0.1:6379> SPOP lottery # 随机抽一个
"user:3"
127.0.0.1:6379> SPOP lottery
"user:1"
4.3 常见场景
| 场景 | 用到的命令 | 说明 |
|---|---|---|
| 文章标签 | SADD/SMEMBERS | 一个文章多个标签 |
| 共同好友/共同关注 | SINTER | 交集运算，一句命令 |
| 推荐关注（你可能认识） | SDIFF | 差集：他有我没有的朋友 |
| 用户标签体系 | SUNION | 并集：覆盖多种特征的用户群 |
| 抽奖/随机推荐 | SRANDMEMBER/SPOP | 随机取元素 |
| 点赞用户集合 | SADD/SREM/SCARD | 记录谁点了赞，快速去重 |
五、ZSet — 有序集合
5.1 为什么需要 ZSet？
Set 是无序的——你能存”用户 A 是 vip”，但不能存”用户 A 消费了 5200 元，排名第 3”。ZSet 给每个元素绑定一个 score（分数），按分数自动排序。
Set 是微信群（只知道有谁），ZSet 是成绩排名表——每个人后面跟着一个分数，按分数从高到低排。
5.2 核心命令
| 命令 | 作用 | 示例 |
|---|---|---|
ZADD key score member | 添加元素（有则更新分数） | ZADD rank 9800 "张三" |
ZREM key member | 删除元素 | ZREM rank "张三" |
ZSCORE key member | 查某个元素的分数 | ZSCORE rank "张三" |
ZRANK key member | 查排名（从小到大，0 开始） | ZRANK rank "张三" |
ZREVRANK key member | 查排名（从大到小） | ZREVRANK rank "张三" |
ZRANGE key start stop | 按排名范围取（升序） | ZRANGE rank 0 2（前三名） |
ZREVRANGE key start stop | 按排名范围取（降序） | ZREVRANGE rank 0 2 |
ZRANGE ... WITHSCORES | 取值时带上分数 | ZREVRANGE rank 0 2 WITHSCORES |
ZRANGEBYSCORE key min max | 按分数范围取 | ZRANGEBYSCORE rank 5000 10000 |
ZCARD key | 元素个数 | ZCARD rank |
ZCOUNT key min max | 分数在范围内的个数 | ZCOUNT rank 5000 10000 |
ZINCRBY key N member | 分数 +N | ZINCRBY rank 100 "张三" |
ZREMRANGEBYRANK key start stop | 按排名删除 | ZREMRANGEBYRANK rank 0 9（删 Top10） |
# 添加玩家 + 积分
127.0.0.1:6379> ZADD game:rank 9800 "张三" 7200 "李四" 5100 "王五" 8900 "赵六"
(integer) 4
# 查看 Top3（从大到小）
127.0.0.1:6379> ZREVRANGE game:rank 0 2 WITHSCORES
1) "张三"
2) "9800"
3) "赵六"
4) "8900"
5) "李四"
6) "7200"
# 查看张三的排名（从大到小）
127.0.0.1:6379> ZREVRANK game:rank "张三"
(integer) 0
# 查看王五的排名和分数
127.0.0.1:6379> ZREVRANK game:rank "王五"
(integer) 3
127.0.0.1:6379> ZSCORE game:rank "王五"
"5100"
# 张三 +100 分
127.0.0.1:6379> ZINCRBY game:rank 100 "张三"
"9900"
# 分数在 5000~10000 的有多少人
127.0.0.1:6379> ZCOUNT game:rank 5000 10000
(integer) 4
# score 存执行时间的时间戳，消费者取 score 最小的
127.0.0.1:6379> ZADD delay:queue 1719000000 "send_email:1001"
(integer) 1
127.0.0.1:6379> ZADD delay:queue 1719000500 "send_sms:1002"
(integer) 1
# 取"现在之前该执行"的任务
127.0.0.1:6379> ZRANGEBYSCORE delay:queue 0 1719000300
1) "send_email:1001"
5.3 常见场景
| 场景 | 用到的命令 | 说明 |
|---|---|---|
| 游戏排行榜 | ZADD/ZREVRANGE/ZINCRBY | 积分更新 + 排名查询 |
| 热搜榜单 | ZINCRBY/ZREVRANGE | 每搜索一次 score +1 |
| 延迟队列 | ZADD/ZRANGEBYSCORE | score = 执行时间戳 |
| 带权重的推荐 | ZADD/ZREVRANGE | score = 推荐权重 |
| 时间线排序 | ZADD/ZREVRANGE | score = 发布时间戳 |
速查表
各数据结构一句话 + 核心命令
| 类型 | 一句话 | 增 | 查 | 删 | 特色 |
|---|---|---|---|---|---|
| String | 一个 key 一个值 | SET | GET/MGET | DEL | INCR 原子自增 |
| Hash | key 下面有多个 field-value | HSET | HGET/HMGET/HGETALL | HDEL | HINCRBY 字段自增 |
| List | 有序可重复的双端列表 | LPUSH/RPUSH | LRANGE/LINDEX | LPOP/RPOP | BLPOP 阻塞弹出 |
| Set | 无序不重复的集合 | SADD | SMEMBERS/SISMEMBER | SREM/SPOP | SINTER/SUNION/SDIFF |
| ZSet | 带分数的有序集合 | ZADD | ZRANGE/ZSCORE/ZRANK | ZREM | ZINCRBY 改分值 |
场景选型速查
| 你要做什么 | 用哪个类型 | 为什么 |
|---|---|---|
| 缓存一个 API 返回的 JSON | String | 直接序列化存，最简单 |
| 缓存用户信息（按字段读写） | Hash | 不用整体序列化/反序列化 |
| 计数器 / 限流 | String + INCR | 原子操作，天生并发安全 |
| 消息队列（轻量） | List | 双端操作，BLPOP 阻塞等 |
| 最新动态 Top N | List + LTRIM | 左边塞新的，裁剪保留 N 条 |
| 去重 / 标签 / 点赞用户 | Set | 自动去重，支持交并差 |
| 共同好友 / 共同关注 | Set + SINTER | 一句命令出结果 |
| 排行榜 / 热搜 | ZSet | score 自动排序 |
| 延迟队列 | ZSet | score = 执行时间，按范围取 |
| 抽奖 / 随机推荐 | Set + SRANDMEMBER | 随机取元素 |
缓存穿透 / 击穿 / 雪崩
缓存穿透（Cache Penetration）
是什么：查询一个根本不存在的数据，缓存里没有，数据库里也没有。每次请求都穿透缓存打到数据库。
用户查 id=99999 的商品（不存在）
→ Redis: 没有
→ MySQL: 也没有
→ 返回空
恶意攻击者用脚本循环查 id=-1, -2, -3...（全不存在）
→ 每次都打穿 Redis 直击 MySQL
→ 数据库被打垮
有人不停按你家门铃问”王大爷在吗”——但小区根本没这个人。你每次都要开门看一眼再回答不在。
解决方案：
| 方案 | 做法 | 优缺点 |
|---|---|---|
| 缓存空值 | 查不到的数据也缓存，value 设 null 或特殊标记，过期时间设短（1-5 分钟） | 简单；浪费一点内存 |
| 布隆过滤器 | 在 Redis 前面加一个布隆过滤器，先判断 key 是否”可能存在” | 高效；实现复杂，有误判率 |
# 缓存空值示例：查不到的商品也存，但短时间过期
SET product:99999 "" EX 60
# 下次查同一个不存在 ID → Redis 返回空 → 不查 MySQL
布隆过滤器
一个很小的数据结构，能告诉你”这个 key 肯定不存在”或”可能存在”。就像图书馆门口的检索电脑——查不到的书名肯定没有，查到了再进库（Redis 缓存/MySQL）。Redis 4.0+ 有内置的布隆过滤器模块
RedisBloom。
缓存击穿（Cache Breakdown）
是什么：一个热点 key（比如双十一某爆款商品）在过期的一瞬间，大量请求同时打到数据库——缓存被”击穿”了一个洞。
某热门商品的缓存 key 在 12:00:00 过期
→ 12:00:00.001，1000 个请求同时到达
→ Redis 返回 nil（过期了）
→ 1000 个请求全部涌向 MySQL 查同一条数据
→ 数据库瞬时压力巨大
演唱会门口只有一个检票口——开门瞬间所有人挤进去，检票员（数据库）被冲垮。
和穿透的区别：穿透是查”不存在的数据”；击穿是查”存在但缓存刚好过期了的热点数据”。
解决方案：
| 方案 | 做法 | 优缺点 |
|---|---|---|
| 互斥锁 | 第一个没命中的请求去查数据库，同时加锁；其他请求等锁释放后从缓存拿 | 保证只有一个请求查 DB；增加复杂度 |
| 永不过期 | 热点数据不设过期时间，用异步任务更新 | 简单；可能返回旧数据（看业务容忍度） |
互斥锁实现
def get_product(product_id: int):
# 1. 先查 Redis
data = redis.get(f"product:{product_id}")
if data:
return data
# 2. 没命中 → 尝试获取锁
lock_key = f"lock:product:{product_id}"
if redis.setnx(lock_key, "1"):
redis.expire(lock_key, 10)
try:
data = mysql.query(product_id)
redis.setex(f"product:{product_id}", 3600, data)
finally:
redis.delete(lock_key)
return data
# 3. 没拿到锁 → 等一会再查 Redis
time.sleep(0.1)
return get_product(product_id)
永不过期方案
核心思路：热点 key 不设 EXPIRE——缓存永远不会因为 TTL 到期而消失。数据更新靠后台异步任务定期从 MySQL 刷新到 Redis。应用层永远只读 Redis，不直接查 MySQL。
请求 → 读 Redis → 永远命中（key 没有 TTL） → 返回
↑
后台定时任务：每 N 秒查 MySQL → 更新 Redis
方式 1：定时任务全量刷新
热点数据量不大（几百到几千条），对实时性要求不高（秒级延迟可接受）。
import asyncio
import json
HOT_PRODUCT_IDS = [1001, 1002, 1003, 1050, 1088]
async def refresh_hot_products():
while True:
for pid in HOT_PRODUCT_IDS:
product = await db.query("SELECT * FROM products WHERE id = %s", pid)
if product:
await redis.set(f"product:{pid}", json.dumps(product))
await asyncio.sleep(30)
async def get_product(product_id: int):
data = await redis.get(f"product:{product_id}")
if data:
return json.loads(data)
return await get_product_from_db(product_id)
实现最简单，但热点列表要手动维护，新增热点需要改代码或配置。
方式 2：逻辑过期时间（推荐）
物理上永不过期（不设 EXPIRE），逻辑上把过期时间写在 value 里。读取时判断——如果”逻辑过期”了，先返回旧数据，再异步触发更新。
Redis 里存的数据结构（value 里包一层）：
{
"data": { "id": 1, "name": "张三", ... },
"expire_at": 1719100000
}
import json
import time
import threading
def get_product_logical_expire(product_id: int):
cache_key = f"product:{product_id}"
lock_key = f"lock:product:{product_id}"
raw = redis.get(cache_key)
if raw is None:
return get_product_from_db(product_id)
payload = json.loads(raw)
data = payload["data"]
expire_at = payload["expire_at"]
if time.time() < expire_at:
return data
# 逻辑过期 → 先返回旧数据，再异步更新
if redis.setnx(lock_key, "1"):
redis.expire(lock_key, 10)
threading.Thread(
target=_refresh_cache, args=(product_id, cache_key, lock_key)
).start()
return data
def _refresh_cache(product_id: int, cache_key: str, lock_key: str):
try:
product = db.query(f"SELECT * FROM products WHERE id = {product_id}")
if product:
payload = {
"data": product,
"expire_at": time.time() + 3600,
}
redis.set(cache_key, json.dumps(payload))
finally:
redis.delete(lock_key)
关键细节：
-
“逻辑过期先返回旧数据再异步更新”——用户请求不会被阻塞
-
互斥锁保证只有一个线程去查 MySQL
-
预热很重要：系统启动时必须把热点数据加载进 Redis，否则第一个请求会穿透
方式 3：Redis 键空间通知 + 定时巡检
更自动化地发现”哪些 key 该刷新了”。
import time
import json
async def get_product_with_tracking(product_id: int):
cache_key = f"product:{product_id}"
raw = await redis.get(cache_key)
if raw:
await redis.zincrby("hotkey:access:count", 1, cache_key)
return json.loads(raw)
return await get_product_from_db(product_id)
async def refresh_stale_hotkeys():
while True:
hot_keys = await redis.zrevrange("hotkey:access:count", 0, 99, withscores=True)
for cache_key, _ in hot_keys:
last_refresh = await redis.hget("hotkey:meta", cache_key)
last_refresh = float(last_refresh) if last_refresh else 0
if time.time() - last_refresh > 300:
product_id = int(cache_key.split(":")[1])
product = await db.query("SELECT * FROM products WHERE id = %s", product_id)
if product:
await redis.set(cache_key, json.dumps(product))
await redis.hset("hotkey:meta", cache_key, time.time())
await asyncio.sleep(60)
三种方案对比
| 方案 | 实现难度 | 实时性 | 维护成本 | 适用场景 |
|---|---|---|---|---|
| 定时全量刷新 | ⭐ 最简单 | 秒级 | 需手动维护热点列表 | 热点数据明确且稳定 |
| 逻辑过期时间 | ⭐⭐ 中等 | 读请求触发，近乎实时 | 低，自动识别过期 | 大多数业务场景 |
| 定时巡检 | ⭐⭐⭐ 较复杂 | 分钟级 | 需额外存储元数据 | 热点 key 动态变化 |
大部分场景**方式 2（逻辑过期时间）**就够了：缓存物理上永不过期 → 彻底消灭击穿；逻辑过期后返回旧数据 + 异步更新 → 用户请求永远快速响应。缺点是可能返回几秒的旧数据——99% 的业务场景都能接受。如果数据一致性要求极高（如库存扣减、余额查询），那不应该用缓存，直接查 MySQL。缓存本质上就是”用一致性换性能”。
缓存雪崩（Cache Avalanche）
是什么：大量 key 在同一时间过期，或者 Redis 挂了，所有请求直接砸向 MySQL。
场景 1：批量设了 1000 个 key，全部 1 小时后过期
→ 1 小时后瞬间全部消失
→ 所有请求打向数据库
场景 2：Redis 进程崩溃 / 重启
→ 缓存全没
→ 数据库承受全部流量
雪崩不是一块雪掉下来（击穿），而是整片山坡的雪同时塌下来——所有缓存一起失效。
解决方案：
| 方案 | 做法 | 优缺点 |
|---|---|---|
| 随机过期时间 | 同一批 key 的过期时间加随机偏移（如 3600 + random(0, 600)） | 最简单有效，避免同时过期 |
| 多级缓存 | 本地缓存（应用内存）+ Redis + MySQL | Redis 挂了还有本地缓存兜底 |
| 限流降级 | 数据库前加限流，请求太多时直接返回降级页面 | 牺牲一部分用户体验保数据库 |
| Redis 高可用 | 主从 + 哨兵 / 集群 | Redis 本身不挂是根本 |
# 随机过期时间示例
SETEX product:1 3700 # 3600 + random(0, 600)
SETEX product:2 3850
SETEX product:3 3550
# 过期时间散开了，不会同时失效
三兄弟对比
| 缓存穿透 | 缓存击穿 | 缓存雪崩 | |
|---|---|---|---|
| 通俗记忆 | 查不存在的 | 热点过期了 | 大片一起过期 |
| 数据在 DB | ❌ 不存在 | ✅ 存在 | ✅ 存在 |
| 范围 | 单个/多个不存在 key | 单个热点 key | 大量 key |
| 核心解法 | 布隆过滤器 / 缓存空值 | 互斥锁 / 永不过期 | 随机过期时间 / 多级缓存 |
常见踩坑
1. KEYS * 在生产环境用
# ❌ 生产禁用！KEYS 会阻塞 Redis，遍历所有 key
KEYS *
# ✅ 用 SCAN 分批迭代，不阻塞
SCAN 0 MATCH user:* COUNT 100
# 返回游标 + 一批 key，下次用新游标继续
KEYS 是 O(n) 且阻塞——有几百万个 key 时就卡几秒，期间所有请求等着。
2. key 命名混乱
# ❌ 毫无命名规范
SET name "张三"
SET 1 "value"
# ✅ 用冒号分层（业界惯例）
SET user:1001:name "张三"
SET product:2001:stock "50"
SET cache:api:userlist "{...}"
3. 不设过期时间
# ❌ 永不过期 → 内存越占越多 → OOM
SET cache:data "{...}"
# ✅ 始终设过期时间（除非确实需要永久）
SETEX cache:data 3600 "{...}"
4. 一个 key 存太大（Big Key）
# ❌ 一个 key 存几 MB 数据
SET user:all "[{...一万个用户...}]"
# ✅ 拆成多个小 key
HSET user:1 name "张三" age 25
HSET user:2 name "李四" age 30
大 key 会让 Redis 单线程卡在序列化/网络传输上，拖慢所有请求。
5. 热 key 没做保护
一个 key 被几十万 QPS 访问 → 单节点 CPU 飙升。解法：本地缓存（应用内存里也存一份），或者key 拆分（hotkey:1、hotkey:2、hotkey:3 随机取一个）。
6. DEL 大 key 时阻塞
# ❌ DEL 大集合 → 阻塞
DEL big:set
# ✅ 用 UNLINK（异步删除，Redis 4.0+）
UNLINK big:set
7. List 当消息队列丢消息
LPOP 弹出后客户端崩了 → 消息没了。生产级消息队列需要 ack 确认机制，Redis List 没有。用 Redis Stream（5.0+）或 RabbitMQ/Kafka。
前置知识补充
json 模块和 asyncio 异步编程——理解缓存击穿代码中用到的 Python 知识。
json 模块：Python 对象 ⇄ 字符串
为什么需要它？ Redis 的 String 类型只能存字符串。但你的业务数据是 Python 字典/列表——需要”翻译”成字符串才能存进去，取出来再”翻译”回来。
flowchart LR Dict["Python dict<br/>{'name':'张三','age':25}"] String["Redis String<br/>{'name':'张三','age':25}"] Dict -- dumps --> String String -- loads --> Dict
核心 4 个函数：
| 函数 | 方向 | 说明 | 示例 |
|---|---|---|---|
json.dumps(obj) | Python → 字符串 | 序列化 | json.dumps({"a":1}) → '{"a":1}' |
json.loads(s) | 字符串 → Python | 反序列化 | json.loads('{"a":1}') → {"a":1} |
json.dump(obj, f) | Python → 文件 | 写入文件 | json.dump(data, open("a.json","w")) |
json.load(f) | 文件 → Python | 从文件读取 | json.load(open("a.json")) |
记忆技巧：带 s 的是处理字符串，不带 s 的是处理文件。
import json
user = {"name": "张三", "age": 25, "tags": ["Python", "Redis"]}
# 序列化存 Redis
json_str = json.dumps(user)
redis.set("user:1", json.dumps({"name": "张三", "age": 25}))
# 反序列化
data = json.loads(redis.get("user:1"))
常见踩坑：
# ❌ 不是所有 Python 对象都能序列化
json.dumps(datetime.now()) # TypeError
# ✅ 需要先转成字符串
json.dumps({"time": datetime.now().isoformat()})
# ❌ loads 接收的字符串必须用双引号
json.loads("{'name': '张三'}") # 报错！
# ✅ 双引号
json.loads('{"name": "张三"}')
asyncio：Python 的异步编程
为什么需要它？ 在等待 IO 的时候，CPU 可以去做别的事。
flowchart TD subgraph 同步代码 A1["查 MySQL 3ms"] --> Wait["等待..."] end subgraph 异步代码 B1["查 MySQL 3ms"] --> B2[等待时做别的] B2 --> B3["查 Redis 0.5ms"] B3 --> B4["发 HTTP 2ms"] end
核心语法：
import asyncio
async def fetch_data(url: str):
print(f"开始请求 {url}")
await asyncio.sleep(2)
print(f"完成请求 {url}")
return f"{url} 的数据"
async def main():
# 顺序执行
r1 = await fetch_data("url1")
# 并发执行：3 个 URL 同时请求
results = await asyncio.gather(
fetch_data("url1"),
fetch_data("url2"),
fetch_data("url3"),
)
print(results)
asyncio.run(main())
三个最常用的并发工具：
| 工具 | 用途 | 示例 |
|---|---|---|
asyncio.gather(*tasks) | 并发执行多个任务，等全部完成 | await asyncio.gather(task1(), task2()) |
asyncio.create_task(coro) | 创建后台任务，不等它完成 | task = asyncio.create_task(update_cache()) |
asyncio.sleep(seconds) | 异步等待（不阻塞其他任务） | await asyncio.sleep(5) |
实战：异步刷新缓存
async def refresh_one_product(pid: int):
product = await db.query("SELECT * FROM products WHERE id = %s", pid)
if product:
await redis.set(f"product:{pid}", json.dumps(product))
async def refresh_all_hot_products():
while True:
hot_ids = [1001, 1002, 1003, 1050, 1088]
await asyncio.gather(*[refresh_one_product(pid) for pid in hot_ids])
await asyncio.sleep(30)
asyncio.run(refresh_all_hot_products())
关键概念：
| 概念 | 一句话解释 |
|---|---|
async def | 定义一个”可以暂停”的函数 |
await | 暂停当前函数，让出 CPU，等这个操作完成再回来 |
| 协程 (coroutine) | 就是 async def 函数的返回值，一个”可暂停的任务” |
| 事件循环 (event loop) | 调度器——管理所有协程，决定谁先跑 |
asyncio.gather | 你们几个一起跑，我在这等你们全部跑完 |
asyncio vs threading vs multiprocessing：
| asyncio（协程） | threading（线程） | multiprocessing（进程） | |
|---|---|---|---|
| 切换方式 | 代码主动 await 让出 | 操作系统强制切换 | 操作系统强制切换 |
| 并发量 | 上万没问题 | 几百就卡了 | 几十个 |
| 适合场景 | IO 密集型（网络、数据库） | IO 密集型 | CPU 密集型 |
| GIL 影响 | 不受影响 | 受 GIL 限制 | 不受影响 |
| 是否有锁问题 | 单线程，无竞态 | 需要锁 | 需要跨进程通信 |
Warning
Python 因为有 GIL，多线程不能用来加速 CPU 密集任务。CPU 密集必须用
multiprocessing（多进程）。asyncio 适用于”同时发 100 个 Redis 查询”或”定时任务并发刷新缓存”场景。
常见踩坑：
# ❌ 在 async 函数里用同步 sleep → 整个事件循环卡死
async def bad():
time.sleep(5) # 阻塞整个线程！
# ✅ 用异步版本
async def good():
await asyncio.sleep(5)
# ❌ 忘记 await → 根本没执行
async def bug():
refresh_cache() # 返回 coroutine 对象，但没执行！
# ✅ 必须 await
async def fixed():
await refresh_cache()
# ❌ 用了同步 Redis 库在 async 函数里
import redis
r = redis.Redis()
async def bad():
r.get("key") # 阻塞！
# ✅ 用异步 Redis 客户端
from redis.asyncio import Redis
r = Redis()
async def good():
await r.get("key")
综合实战
五种数据结构命令实操
在 redis-cli 中完成以下所有操作。
练习 1：String — 文章阅读量计数器
# 1. 初始化三篇文章的阅读量
MSET article:1:views 0 article:2:views 0 article:3:views 0
# 2. 文章 1 被阅读 5 次，文章 2 被阅读 3 次
INCRBY article:1:views 5
INCRBY article:2:views 3
# 3. 文章 2 又被阅读 1 次
INCR article:2:views
# 4. 查看三篇文章的阅读量
MGET article:1:views article:2:views article:3:views
# 5. 设置一个验证码，300 秒后过期
SETEX verify:13800138000 300 "123456"
# 6. 检查验证码还存在
TTL verify:13800138000
# 期望: (integer) 295 左右
练习 2：Hash — 用户信息缓存
# 1. 存入用户 1001 的信息
HSET user:1001 name "张三" age 25 email "zhangsan@qq.com" city "北京"
# 2. 存入用户 1002 的信息
HSET user:1002 name "李四" age 30 email "lisi@qq.com" city "上海"
# 3. 只查询用户 1001 的 name 和 email
HMGET user:1001 name email
# 4. 修改用户 1001 的 city 为"深圳"
HSET user:1001 city "深圳"
# 5. 用户 1001 过生日，年龄 +1
HINCRBY user:1001 age 1
# 6. 查看用户 1001 所有信息
HGETALL user:1001
# 7. 判断用户 1002 有没有 phone 字段
HEXISTS user:1002 phone
# 期望: (integer) 0
练习 3：List — 最新动态 + 消息队列
# 1. 用户 1001 发了 5 条动态
LPUSH timeline:1001 "今天天气真好" "Redis 真好用" "中午吃啥" "学会了 List" "晚安"
# 2. 只保留最近 3 条
LTRIM timeline:1001 0 2
# 3. 查看 timeline
LRANGE timeline:1001 0 -1
# 4. 生产者塞 3 个任务
LPUSH task:queue "send_email:1001" "send_sms:1002" "compress_img:1003"
# 5. 消费者取出并处理
RPOP task:queue # → "send_email:1001"
RPOP task:queue # → "send_sms:1002"
# 6. 还剩几个任务？
LLEN task:queue # → (integer) 1
# 7. 阻塞等待新任务（5 秒超时）
BLPOP task:queue 5
# 期望: 1) "task:queue" 2) "compress_img:1003"
练习 4：Set — 标签 + 共同好友
# 1. 给文章打标签
SADD article:1:tags "Python" "Redis" "后端"
SADD article:2:tags "Python" "Django" "后端"
SADD article:3:tags "JavaScript" "React" "前端"
# 2. 查文章 1 的所有标签
SMEMBERS article:1:tags
# 3. 文章 1 和文章 2 的共同标签
SINTER article:1:tags article:2:tags
# 4. 文章 1 和文章 3 有共同标签吗？
SINTER article:1:tags article:3:tags
# 5. 我的好友
SADD friends:me "Alice" "Bob" "Charlie" "David"
# 6. Bob 的好友
SADD friends:Bob "Alice" "David" "Eve" "Frank"
# 7. 共同好友
SINTER friends:me friends:Bob
# 8. 推荐好友——Bob 有但我没有的
SDIFF friends:Bob friends:me
# 9. 所有我们认识的人（并集）
SUNION friends:me friends:Bob
# 期望: Alice、Bob、Charlie、David、Eve、Frank
练习 5：ZSet — 游戏排行榜 + 延迟队列
# 1. 录入玩家积分
ZADD game:score 9800 "player:A" 7200 "player:B" 5100 "player:C" 8900 "player:D" 6500 "player:E"
# 2. 查看 Top3
ZREVRANGE game:score 0 2 WITHSCORES
# 3. 查看 player:C 的排名和分数
ZREVRANK game:score "player:C"
ZSCORE game:score "player:C"
# 4. player:C 完成一局游戏 +300 分
ZINCRBY game:score 300 "player:C"
# 5. 分数范围统计
ZCOUNT game:score 5000 10000
# 6. 淘汰倒数 2 名
ZREMRANGEBYRANK game:score 0 1
# 7. 查看剩下的
ZREVRANGE game:score 0 -1 WITHSCORES
# 8. 添加延迟任务
ZADD delay:task 1719100000 "send_reward:user:1001"
ZADD delay:task 1719100100 "send_reward:user:1002"
ZADD delay:task 1719105000 "send_reward:user:1003"
# 9. 取该执行的任务
ZRANGEBYSCORE delay:task 0 1719100300
# 期望: user:1001、user:1002
缓存场景：用户信息查询
前置：
pip install redis，确保 Redis 已启动（redis-cli ping返回 PONG）。数据库用 Python 内置的 sqlite3。
"""
可运行的缓存查询示例：
请求 → 查 Redis（缓存）→ 命中 → 返回
↓ 没命中
查 SQLite（数据库）→ 写回 Redis → 返回
"""
import sqlite3
import json
import time
import redis
# ===== 1. 初始化 Redis 连接 =====
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# ===== 2. 初始化 SQLite 并插入测试数据 =====
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE users (id INT, name TEXT, age INT, email TEXT)")
db.execute("INSERT INTO users VALUES (1, '张三', 25, 'zs@qq.com')")
db.execute("INSERT INTO users VALUES (2, '李四', 30, 'ls@qq.com')")
db.commit()
# ===== 3. 带缓存的查询函数 =====
def get_user(user_id: int) -> dict | None:
cache_key = f"user:{user_id}"
# 第 1 步：查 Redis
cached = r.get(cache_key)
if cached:
print(f" ✅ 缓存命中")
return json.loads(cached)
# 第 2 步：没命中，查数据库
print(f" ❌ 缓存未命中 → 查数据库")
row = db.execute(
"SELECT id, name, age, email FROM users WHERE id = ?", (user_id,)
).fetchone()
if not row:
return None
user = {"id": row[0], "name": row[1], "age": row[2], "email": row[3]}
# 第 3 步：写回 Redis
r.setex(cache_key, 3600, json.dumps(user))
print(f" 📝 已写入缓存 (TTL=3600s)")
return user
# ===== 4. 测试 =====
print("第 1 次查询 user:1：")
t1 = time.time()
user = get_user(1)
print(f" 结果: {user} | 耗时: {(time.time() - t1)*1000:.1f}ms\n")
print("第 2 次查询 user:1：")
t2 = time.time()
user = get_user(1)
print(f" 结果: {user} | 耗时: {(time.time() - t2)*1000:.1f}ms\n")
print("查一个不存在的用户 (id=999)：")
user = get_user(999)
print(f" 结果: {user}")
运行输出：
第 1 次查询 user:1：
❌ 缓存未命中 → 查数据库
📝 已写入缓存 (TTL=3600s)
结果: {'id': 1, 'name': '张三', 'age': 25, 'email': 'zs@qq.com'} | 耗时: 3.2ms
第 2 次查询 user:1：
✅ 缓存命中
结果: {'id': 1, 'name': '张三', 'age': 25, 'email': 'zs@qq.com'} | 耗时: 0.8ms
查一个不存在的用户 (id=999)：
❌ 缓存未命中 → 查数据库
结果: None
| 你看到的 | 说明 |
|---|---|
| 第 1 次慢（3.2ms） | 走了数据库 + 写缓存 |
| 第 2 次快（0.8ms） | 直接命中缓存，没碰数据库 |
| 不存在的用户返回 None | 没有缓存空值（简化处理），实际项目参考缓存穿透节 |
速记卡（面试闪卡）
Q1：一句话讲清「Redis 数据结构与缓存」到底是什么？
A：这篇讲 Redis 为什么快、5 种数据结构怎么用，以及缓存三大坑（穿透/击穿/雪崩）怎么防。
Q2：一、Redis 是什么（为什么快）—— 怎么理解？ —— 怎么理解？
A：MySQL 是仓库（空间大、走磁盘慢），Redis 是工作台工具箱（纯内存、伸手就拿）。快的四点：纯内存、单线程无锁竞争、IO 多路复用、简单 key-value。量级上 Redis QPS 10 万+，比 MySQL 快两个数量级——瓶颈不在 CPU 在带宽。
Q3：二、五种数据结构与命令 —— 怎么理解？ —— 怎么理解？
A：String 计数/分布式锁（INCR/SETNX）；Hash 按字段读写对象；List 双端做队列（BLPOP 阻塞）；Set 交并差做共同好友；ZSet 跳表做排行榜。选型口诀：缓存 JSON 用 String、按字段用 Hash、队列用 List、去重交集用 Set、排行榜用 ZSet。
Q4：三、缓存穿透 / 击穿 / 雪崩 —— 怎么理解？ —— 怎么理解？
A：穿透=查不存在的数据，用布隆过滤器或缓存空值；击穿=热点 key 过期瞬间请求涌向 DB，用互斥锁或逻辑过期（物理不设 TTL，value 里写过期时间，过期先返旧数据再异步更新）；雪崩=大量 key 同时过期，用随机过期时间+多级缓存。
Q5：四、常见踩坑 —— 怎么理解？ —— 怎么理解？
A：①生产禁用 KEYS *（O(n) 阻塞，用 SCAN）；②key 用冒号分层命名；③永远设过期时间（否则 OOM）；④避免 Big Key（拆小 key）；⑤热 key 做本地缓存或拆分；⑥删大 key 用 UNLINK 别 DEL；⑦别用 List 当正经消息队列（没 ACK）。
Q6：核心速记主线有哪些？
-
Redis 快：纯内存+单线程+IO 多路复用+简单结构
-
五种结构：String/List/Hash/Set/ZSet 各司其职
-
缓存三兄弟：穿透/击穿/雪崩 及各自解法
-
7 个踩坑：KEYS 禁用、大 key 拆、UNLINK 删等
口诀
A：内存单线多路通，
Redis 比 MySQL 快；
三兄弟坑要记牢，
KEYS 禁用 SCAN 代。
相关链接
- 对应理论（概念详解）：Redis 五种数据结构（理论篇）