> ## Content Index > Fetch the complete content index at: https://blog.vercanti.com/llms.txt > Use this file to discover other available public pages before exploring further. # Redis 完全指南 - URL: https://blog.vercanti.com/redis-wan-quan-zhi-nan/ - Published: 2026-08-28T14:35:38.000Z - Updated: 2026-08-28T14:59:14.000Z - Description: 最后更新：2026-03-05 1. Redis完全指南 · 基础概念(/redis-wan-quan-zhi-nan/#%E5%9F%BA%E7%A1%80%E6%A6%82%E5%BF%B5) 2. Redis完全指南 · String（字符串）(/redis-wan-quan-zhi-nan/#string%EF%BC%88%E5%AD%97%E7%AC%A6%E4%B8%B2%EF%BC%89) 3. Redis完全指南 · List（列表）(/redis-wan-quan-zhi-nan/#list%EF%BC%88%E5%88%97%E8%A - Author: yellowdog - Tags: 数据库 最后更新：2026-03-05 > 官方文档： > redis-py 文档： --- ## 目录 1. [Redis完全指南 · 基础概念](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E5%9F%BA%E7%A1%80%E6%A6%82%E5%BF%B5) 2. [Redis完全指南 · String（字符串）](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#string%EF%BC%88%E5%AD%97%E7%AC%A6%E4%B8%B2%EF%BC%89) 3. [Redis完全指南 · List（列表）](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#list%EF%BC%88%E5%88%97%E8%A1%A8%EF%BC%89) 4. [Redis完全指南 · Hash（哈希）](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#hash%EF%BC%88%E5%93%88%E5%B8%8C%EF%BC%89) 5. [Redis完全指南 · Set（集合）](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#set%EF%BC%88%E9%9B%86%E5%90%88%EF%BC%89) 6. [Redis完全指南 · ZSet（有序集合）](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#zset%EF%BC%88%E6%9C%89%E5%BA%8F%E9%9B%86%E5%90%88%EF%BC%89) 7. [Redis完全指南 · 通用键命令](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E9%80%9A%E7%94%A8%E9%94%AE%E5%91%BD%E4%BB%A4) 8. [Redis完全指南 · Python 集成：redis-py](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#python-%E9%9B%86%E6%88%90%EF%BC%9Aredis-py) 9. [Redis完全指南 · 持久化：RDB 与 AOF](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E6%8C%81%E4%B9%85%E5%8C%96%EF%BC%9Ardb-%E4%B8%8E-aof) 10. [Redis完全指南 · 发布订阅](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E5%8F%91%E5%B8%83%E8%AE%A2%E9%98%85) 11. [Redis完全指南 · 过期策略与内存淘汰](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E8%BF%87%E6%9C%9F%E7%AD%96%E7%95%A5%E4%B8%8E%E5%86%85%E5%AD%98%E6%B7%98%E6%B1%B0) 12. [Redis完全指南 · 分布式锁](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81) 13. [Redis完全指南 · 最佳实践](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E6%9C%80%E4%BD%B3%E5%AE%9E%E8%B7%B5) 14. [Redis完全指南 · 常见陷阱与注意事项](https://blog.vercanti.com/redis-wan-quan-zhi-nan/#%E5%B8%B8%E8%A7%81%E9%99%B7%E9%98%B1%E4%B8%8E%E6%B3%A8%E6%84%8F%E4%BA%8B%E9%A1%B9) --- ## 基础概念 Redis（Remote Dictionary Server）是一个开源的、基于内存的数据结构存储系统，可用作数据库、缓存和消息代理。 ### 核心特性 | 特性 | 说明 | | ------ | ---------------------------------------- | | 纯内存存储 | 所有数据存储在内存，读写速度极快（10万+ QPS） | | 单线程模型 | 命令处理单线程，无并发竞争，天然线程安全 | | 持久化支持 | RDB 快照 + AOF 日志，两种方式可组合使用 | | 多种数据结构 | String / List / Hash / Set / ZSet 五大基本类型 | | 原子操作 | 所有命令原子执行，支持 Lua 脚本原子批量执行 | | 发布订阅 | 内置 Pub/Sub 消息系统 | | 集群支持 | Sentinel 高可用 + Cluster 分片扩展 | ### 键命名规范 ``` 业务:对象:ID:字段 示例：user:info:1001:profile order:status:20240101 cache:api:list_users:page_1 ``` 推荐使用冒号 `:` 分隔层级，Obsidian 等工具会将其视为命名空间。键长度建议不超过 1024 字节。 --- ## String（字符串） String 是 Redis 最基本的数据类型，可存储字符串、整数、浮点数、二进制数据（最大 512 MB）。 ### SET 设置键值对。 ``` SET key value [NX | XX] [GET] [EX seconds | PX milliseconds | EXAT unix-time-seconds | PXAT unix-time-milliseconds | KEEPTTL] ``` | 参数 | 类型 | 默认值 | 说明 | | --------------------------- | ------- | --- | -------------------- | | key | string | 必填 | 键名 | | value | string | 必填 | 值 | | NX | flag | 无 | 键不存在时才设置（Not eXists） | | XX | flag | 无 | 键已存在时才设置 | | GET | flag | 无 | 返回旧值（Redis 6.2+） | | EX seconds | integer | 无 | 设置过期时间（秒） | | PX milliseconds | integer | 无 | 设置过期时间（毫秒） | | EXAT unix-time-seconds | integer | 无 | 设置过期的 Unix 时间戳（秒） | | PXAT unix-time-milliseconds | integer | 无 | 设置过期的 Unix 时间戳（毫秒） | | KEEPTTL | flag | 无 | 保留键原有的 TTL | ```bash SET name "Alice" SET counter 100 EX 3600 # 1小时过期 SET lock "1" NX EX 30 # 分布式锁惯用法 SET name "Bob" XX GET # 仅当存在时更新，返回旧值 "Alice" ``` ### GET 获取键的值。键不存在返回 nil。 ``` GET key ``` | 参数 | 类型 | 默认值 | 说明 | | --- | ------ | --- | -- | | key | string | 必填 | 键名 | ```bash GET name # "Alice" GET noexist # (nil) ``` ### MSET / MGET 批量设置/获取，减少网络往返。 ``` MSET key value [key value ...] MGET key [key ...] ``` ```bash MSET a 1 b 2 c 3 MGET a b c noexist # ["1", "2", "3", nil] ``` ### GETSET 设置新值并返回旧值（已废弃，推荐 `SET key value GET`）。 ### INCR / INCRBY / INCRBYFLOAT 原子递增，常用于计数器。值必须为整数类型（或浮点型，针对 INCRBYFLOAT）。 ``` INCR key INCRBY key increment INCRBYFLOAT key increment DECR key DECRBY key decrement ``` | 参数 | 类型 | 默认值 | 说明 | | --------- | ------------- | ---------------------- | ----------- | | key | string | 必填 | 键名，不存在则视为 0 | | increment | integer/float | 必填（INCRBY/INCRBYFLOAT） | 递增量，可为负数 | ```bash INCR page_view # 1（键不存在从0开始） INCRBY page_view 10 # 11 INCRBYFLOAT price 0.5 # 浮点递增 DECR stock # 库存递减 ``` ### APPEND 追加值到已有字符串末尾。 ``` APPEND key value ``` | 参数 | 类型 | 默认值 | 说明 | | ----- | ------ | --- | --------- | | key | string | 必填 | 键名，不存在则创建 | | value | string | 必填 | 追加的内容 | ```bash APPEND log "2024-01-01 login\n" STRLEN log # 返回字节长度 ``` ### GETRANGE / SETRANGE 操作字符串的子串。 ``` GETRANGE key start end SETRANGE key offset value ``` | 参数 | 类型 | 默认值 | 说明 | | ------ | ------- | --- | --------------------- | | key | string | 必填 | 键名 | | start | integer | 必填 | 起始索引（支持负数，-1 为最后一个字符） | | end | integer | 必填 | 结束索引（包含） | | offset | integer | 必填 | 偏移量（字节） | ```bash SET str "Hello World" GETRANGE str 0 4 # "Hello" GETRANGE str -5 -1 # "World" SETRANGE str 6 "Redis" # "Hello Redis" ``` ### SETNX（已废弃） 等价于 `SET key value NX`，推荐使用 SET 的 NX 选项。 --- ## List（列表） List 是一个有序的字符串链表，支持从两端推入/弹出。底层实现为压缩列表（listpack）或双向链表。 适用场景：消息队列、操作日志、最新动态列表。 ### LPUSH / RPUSH 从左端/右端推入一个或多个元素。 ``` LPUSH key element [element ...] RPUSH key element [element ...] ``` | 参数 | 类型 | 默认值 | 说明 | | ------- | ------ | -------- | ------------ | | key | string | 必填 | 键名，不存在则自动创建 | | element | string | 必填（至少一个） | 元素值，多个时按顺序推入 | ```bash RPUSH queue "task1" "task2" "task3" # 右进：[task1, task2, task3] LPUSH stack "a" "b" "c" # 左进：[c, b, a] ``` ### LPOP / RPOP 从左端/右端弹出元素。 ``` LPOP key [count] RPOP key [count] ``` | 参数 | 类型 | 默认值 | 说明 | | ----- | ------- | --- | ---------------- | | key | string | 必填 | 键名 | | count | integer | 1 | 弹出数量（Redis 6.2+） | ```bash LPOP queue # "task1"（消费队列头部） RPOP stack # "a"（弹出栈顶） LPOP queue 2 # 一次弹出2个，返回列表 ``` ### BLPOP / BRPOP 阻塞版弹出，队列为空时阻塞等待，适合实现消费者模式。 ``` BLPOP key [key ...] timeout BRPOP key [key ...] timeout ``` | 参数 | 类型 | 默认值 | 说明 | | ------- | ------ | -------- | ---------------- | | key | string | 必填（至少一个） | 监听的键，支持多个（按顺序检查） | | timeout | float | 必填 | 超时秒数，0 表示永久阻塞 | ```bash BLPOP queue 5 # 阻塞最多5秒等待消息，返回 [键名, 值] BLPOP q1 q2 0 # 永久阻塞，监听多个队列 ``` ### LRANGE 获取列表片段（不修改列表）。 ``` LRANGE key start stop ``` | 参数 | 类型 | 默认值 | 说明 | | ----- | ------- | --- | ---------------- | | key | string | 必填 | 键名 | | start | integer | 必填 | 起始索引（0为头，负数从尾计） | | stop | integer | 必填 | 结束索引（包含，-1为最后一个） | ```bash LRANGE queue 0 -1 # 获取全部 LRANGE queue 0 9 # 获取前10个 LRANGE queue -5 -1 # 获取最后5个 ``` ### LLEN 返回列表长度。 ``` LLEN key ``` ```bash LLEN queue # 队列中的元素数量 ``` ### LINDEX 按索引获取元素，O(n) 复杂度，慎用于大列表。 ``` LINDEX key index ``` | 参数 | 类型 | 默认值 | 说明 | | ----- | ------- | --- | ------- | | key | string | 必填 | 键名 | | index | integer | 必填 | 索引，支持负数 | ### LSET 按索引设置元素值。 ``` LSET key index element ``` ### LINSERT 在指定元素前/后插入新元素，O(n) 复杂度。 ``` LINSERT key BEFORE|AFTER pivot element ``` | 参数 | 类型 | 默认值 | 说明 | | ------------ | ------ | --- | -------------- | | key | string | 必填 | 键名 | | BEFORE/AFTER | flag | 必填 | 插入位置 | | pivot | string | 必填 | 参照元素（从左找第一个匹配） | | element | string | 必填 | 新元素 | ### LREM 删除列表中匹配的元素。 ``` LREM key count element ``` | 参数 | 类型 | 默认值 | 说明 | | ------- | ------- | --- | ---------------------- | | key | string | 必填 | 键名 | | count | integer | 必填 | \>0 从头删 count 个；<0 从尾删 | | element | string | 必填 | 要删除的元素值 | ### LTRIM 裁剪列表，只保留指定范围。 ``` LTRIM key start stop ``` ```bash # 只保留最新的100条 RPUSH logs "entry" LTRIM logs -100 -1 ``` --- ## Hash（哈希） Hash 是一个键值对集合，适合存储对象。底层使用压缩列表（listpack）或哈希表。 适用场景：存储用户信息、商品详情等结构化对象，比 JSON 字符串更灵活（可单字段更新）。 ### HSET 设置一个或多个字段（Redis 4.0+ 支持多字段）。 ``` HSET key field value [field value ...] ``` | 参数 | 类型 | 默认值 | 说明 | | ----- | ------ | --- | ---- | | key | string | 必填 | 哈希键名 | | field | string | 必填 | 字段名 | | value | string | 必填 | 字段值 | ```bash HSET user:1001 name "Alice" age 30 email "alice@example.com" HSET user:1001 age 31 # 更新单个字段 ``` ### HGET / HMGET 获取一个或多个字段值。 ``` HGET key field HMGET key field [field ...] ``` ```bash HGET user:1001 name # "Alice" HMGET user:1001 name age # ["Alice", "31"] ``` ### HGETALL 获取所有字段和值，返回扁平化列表（字段名/值交替）。 ``` HGETALL key ``` ```bash HGETALL user:1001 # 1) "name" # 2) "Alice" # 3) "age" # 4) "31" # 5) "email" # 6) "alice@example.com" ``` 注意：大哈希（字段数量多）调用 HGETALL 会阻塞服务器，建议用 HSCAN 分批获取。 ### HDEL 删除一个或多个字段。 ``` HDEL key field [field ...] ``` ### HEXISTS 判断字段是否存在。 ``` HEXISTS key field ``` 返回 1（存在）或 0（不存在）。 ### HLEN 返回哈希中字段数量。 ``` HLEN key ``` ### HKEYS / HVALS 分别获取所有字段名 / 所有字段值。 ``` HKEYS key HVALS key ``` ### HINCRBY / HINCRBYFLOAT 对哈希中的整数/浮点字段原子递增。 ``` HINCRBY key field increment HINCRBYFLOAT key field increment ``` | 参数 | 类型 | 默认值 | 说明 | | --------- | ------------- | --- | ------------ | | key | string | 必填 | 哈希键名 | | field | string | 必填 | 字段名，不存在则视为 0 | | increment | integer/float | 必填 | 增量，可为负数 | ```bash HINCRBY user:1001 login_count 1 HINCRBYFLOAT user:1001 balance -9.99 ``` ### HSCAN 游标式迭代哈希字段，避免阻塞。 ``` HSCAN key cursor [MATCH pattern] [COUNT count] ``` | 参数 | 类型 | 默认值 | 说明 | | ------------- | ------- | --- | --------------- | | key | string | 必填 | 键名 | | cursor | integer | 必填 | 游标，首次调用传 0 | | MATCH pattern | string | 无 | 字段名匹配模式（glob风格） | | COUNT count | integer | 10 | 每次返回的大致数量（非精确） | ```bash HSCAN user:1001 0 MATCH "addr*" COUNT 100 ``` --- ## Set（集合） Set 是无序、不重复的字符串集合。底层使用 listpack 或哈希表。 适用场景：标签系统、好友关系、抽奖（去重随机选取）、UV 计数。 ### SADD 添加一个或多个元素。 ``` SADD key member [member ...] ``` | 参数 | 类型 | 默认值 | 说明 | | ------ | ------ | -------- | -------- | | key | string | 必填 | 集合键名 | | member | string | 必填（至少一个） | 元素，重复则忽略 | ```bash SADD tags "python" "redis" "backend" SADD online_users "uid_1001" "uid_1002" ``` ### SREM 删除一个或多个元素。 ``` SREM key member [member ...] ``` ### SMEMBERS 获取集合所有元素（无序）。大集合慎用，推荐 SSCAN。 ``` SMEMBERS key ``` ### SISMEMBER / SMISMEMBER 判断元素是否在集合中。 ``` SISMEMBER key member SMISMEMBER key member [member ...] ``` ```bash SISMEMBER tags "python" # 1 SMISMEMBER tags "python" "java" # [1, 0] ``` ### SCARD 返回集合元素数量（基数）。 ``` SCARD key ``` ### SRANDMEMBER 随机返回集合中的元素（不删除）。 ``` SRANDMEMBER key [count] ``` | 参数 | 类型 | 默认值 | 说明 | | ----- | ------- | --- | ---------------------------- | | key | string | 必填 | 键名 | | count | integer | 1 | 正数：返回不重复的 count 个；负数：允许重复，返回 | ```bash SRANDMEMBER prize_pool 3 # 抽3个获奖者（不重复） SRANDMEMBER prize_pool -10 # 允许重复，用于模拟 ``` ### SPOP 随机弹出元素（删除并返回）。 ``` SPOP key [count] ``` ### 集合运算 ``` SUNION key [key ...] # 并集 SINTER key [key ...] # 交集 SDIFF key [key ...] # 差集（第一个集合有，后续集合没有） SUNIONSTORE dest key [key ...] # 并集存入 dest SINTERSTORE dest key [key ...] # 交集存入 dest SDIFFSTORE dest key [key ...] # 差集存入 dest ``` ```bash # 共同好友 SINTER user:1001:friends user:1002:friends # 推荐好友（我有 对方没有） SDIFF user:1001:friends user:1002:friends # 合并两组标签 SUNIONSTORE all_tags tags:article:1 tags:article:2 ``` ### SSCAN 游标式迭代集合。参数同 HSCAN 的 cursor/MATCH/COUNT。 --- ## ZSet（有序集合） ZSet 是带分数（score）的有序集合，元素按 score 从小到大排序，score 为浮点数。底层使用 listpack 或跳表+哈希表。 适用场景：排行榜、带权重的优先级队列、范围查询（如按时间戳排序的时间线）。 ### ZADD 添加或更新元素。 ``` ZADD key [NX | XX] [GT | LT] [CH] [INCR] score member [score member ...] ``` | 参数 | 类型 | 默认值 | 说明 | | ------ | ------ | --- | ------------------------------------- | | key | string | 必填 | 键名 | | NX | flag | 无 | 只添加新元素，不更新已有元素 | | XX | flag | 无 | 只更新已有元素，不添加新元素 | | GT | flag | 无 | 只有新 score 大于当前 score 时才更新（Redis 6.2+） | | LT | flag | 无 | 只有新 score 小于当前 score 时才更新（Redis 6.2+） | | CH | flag | 无 | 返回值改为：新增数量 + score变化数量之和（默认只返回新增数量） | | INCR | flag | 无 | 将 score 作为增量（类似 ZINCRBY） | | score | float | 必填 | 分数，支持 +inf/-inf | | member | string | 必填 | 元素名 | ```bash ZADD leaderboard 9800 "Alice" 8700 "Bob" 9500 "Charlie" ZADD leaderboard NX 9000 "Dave" # 只添加，不更新 ZADD leaderboard GT 10000 "Alice" # 仅当新分数更高时更新 ZADD leaderboard INCR 100 "Alice" # Alice 加 100 分 ``` ### ZRANGE / ZREVRANGE 按索引范围获取元素（Redis 6.2+ ZRANGE 支持 REV/BYSCORE/BYLEX/LIMIT 选项，可替代下述多个命令）。 ``` ZRANGE key min max [BYSCORE | BYLEX] [REV] [LIMIT offset count] [WITHSCORES] ZREVRANGE key start stop [WITHSCORES] # 从高到低，已废弃，推荐 ZRANGE ... REV ``` | 参数 | 类型 | 默认值 | 说明 | | ------------------ | -------------- | --- | ---------------------------------------- | | key | string | 必填 | 键名 | | min | integer/string | 必填 | 起始（索引/分数/字典序） | | max | integer/string | 必填 | 结束（同上） | | BYSCORE | flag | 无 | 按分数范围（min/max 为分数，支持 -inf/+inf，(N 表示开区间） | | BYLEX | flag | 无 | 按字典序范围（分数必须相同） | | REV | flag | 无 | 反转顺序（从高到低） | | LIMIT offset count | integers | 无 | 分页，配合 BYSCORE/BYLEX 使用 | | WITHSCORES | flag | 无 | 同时返回分数 | ```bash # 获取排行榜前10（从高到低） ZRANGE leaderboard 0 9 REV WITHSCORES # 按分数范围查询（旧语法） ZRANGEBYSCORE leaderboard 8000 10000 WITHSCORES LIMIT 0 10 # 新语法等价 ZRANGE leaderboard 8000 10000 BYSCORE WITHSCORES LIMIT 0 10 ``` ### ZRANK / ZREVRANK 获取元素的排名（从0开始）。 ``` ZRANK key member [WITHSCORE] ZREVRANK key member [WITHSCORE] ``` ```bash ZRANK leaderboard "Alice" # 从低到高的排名 ZREVRANK leaderboard "Alice" # 从高到低的排名（第1名返回0） ``` ### ZSCORE 获取元素的分数。 ``` ZSCORE key member ``` ### ZINCRBY 对元素的分数原子递增。 ``` ZINCRBY key increment member ``` ```bash ZINCRBY leaderboard 100 "Alice" # Alice 加分 ``` ### ZREM 删除一个或多个元素。 ``` ZREM key member [member ...] ``` ### ZCARD 返回有序集合元素总数。 ``` ZCARD key ``` ### ZCOUNT 返回分数在指定范围内的元素数量。 ``` ZCOUNT key min max ``` ```bash ZCOUNT leaderboard 9000 +inf # 9000分以上有多少人 ZCOUNT leaderboard -inf +inf # 等同 ZCARD ``` ### ZPOPMIN / ZPOPMAX 弹出分数最小/最大的元素。 ``` ZPOPMIN key [count] ZPOPMAX key [count] ``` ### BZPOPMIN / BZPOPMAX 阻塞版弹出。 ``` BZPOPMIN key [key ...] timeout BZPOPMAX key [key ...] timeout ``` ### ZRANGEBYLEX 按字典序范围获取元素（仅当所有元素 score 相同时有意义）。已由 ZRANGE BYLEX 替代。 ``` ZRANGEBYLEX key min max [LIMIT offset count] ``` min/max 格式：`[value`（包含）、`(value`（不含）、`-`（最小）、`+`（最大）。 ```bash ZADD words 0 "apple" 0 "banana" 0 "cherry" 0 "date" ZRANGEBYLEX words "[b" "[d" # ["banana", "cherry"] ``` ### ZUNIONSTORE / ZINTERSTORE / ZDIFFSTORE 有序集合运算，结果存入新键。 ``` ZUNIONSTORE dest numkeys key [key ...] [WEIGHTS weight ...] [AGGREGATE SUM | MIN | MAX] ZINTERSTORE dest numkeys key [key ...] [WEIGHTS weight ...] [AGGREGATE SUM | MIN | MAX] ZDIFFSTORE dest numkeys key [key ...] ``` | 参数 | 类型 | 默认值 | 说明 | | --------- | ----------- | --- | ----------- | | dest | string | 必填 | 目标键名 | | numkeys | integer | 必填 | 参与运算的键数量 | | WEIGHTS | float list | 全1 | 各键的权重系数 | | AGGREGATE | SUM/MIN/MAX | SUM | 相同元素的分数合并策略 | ```bash # 综合排行：两个榜单加权求和 ZUNIONSTORE combined 2 rank_week rank_history WEIGHTS 2 1 AGGREGATE SUM ``` --- ## 通用键命令 ### DEL 删除一个或多个键。 ``` DEL key [key ...] ``` 返回成功删除的键数量。 ### EXISTS 判断键是否存在。 ``` EXISTS key [key ...] ``` 返回存在的键数量（传入多个键时，同一个键多次传入会多次计数）。 ### EXPIRE / PEXPIRE 设置键的过期时间（秒/毫秒）。 ``` EXPIRE key seconds [NX | XX | GT | LT] PEXPIRE key milliseconds [NX | XX | GT | LT] ``` | 参数 | 类型 | 默认值 | 说明 | | -------------------- | ------- | --- | ------------------------- | | key | string | 必填 | 键名 | | seconds/milliseconds | integer | 必填 | 过期时长 | | NX | flag | 无 | 仅当键没有过期时间时才设置（Redis 7.0+） | | XX | flag | 无 | 仅当键已有过期时间时才设置 | | GT | flag | 无 | 仅当新过期时间大于当前时才设置 | | LT | flag | 无 | 仅当新过期时间小于当前时才设置 | ### EXPIREAT / PEXPIREAT 设置键在某个 Unix 时间戳过期。 ``` EXPIREAT key unix-time-seconds PEXPIREAT key unix-time-milliseconds ``` ### TTL / PTTL 查看键的剩余过期时间（秒/毫秒）。 ``` TTL key PTTL key ``` 返回值：正整数（剩余时间）、-1（永久有效）、-2（键不存在）。 ### PERSIST 移除键的过期时间，使其永久有效。 ``` PERSIST key ``` ### RENAME / RENAMENX 重命名键。RENAMENX 仅当目标键不存在时才重命名。 ``` RENAME key newkey RENAMENX key newkey ``` ### TYPE 返回键的数据类型。 ``` TYPE key ``` 返回：string / list / hash / set / zset / none（不存在）。 ### OBJECT ENCODING 查看键的底层编码（调试用）。 ``` OBJECT ENCODING key ``` | 类型 | 编码 | 触发条件 | | ------ | --------- | -------------------------- | | String | int | 值为整数 | | String | embstr | 值长度 <= 44字节 | | String | raw | 值长度 > 44字节 | | List | listpack | 元素数 <= 128 且每个元素 <= 64字节 | | List | quicklist | 超出上述条件 | | Hash | listpack | 字段数 <= 128 且每个字段/值 <= 64字节 | | Hash | hashtable | 超出上述条件 | | Set | listpack | 元素数 <= 128 且每个元素 <= 64字节 | | Set | intset | 元素全为整数且数量 <= 512 | | Set | hashtable | 超出上述条件 | | ZSet | listpack | 元素数 <= 128 且每个元素 <= 64字节 | | ZSet | skiplist | 超出上述条件 | ### SCAN 游标式迭代所有键，替代 KEYS（KEYS 会阻塞服务器）。 ``` SCAN cursor [MATCH pattern] [COUNT count] [TYPE type] ``` | 参数 | 类型 | 默认值 | 说明 | | ------------- | ------- | --- | ------------------- | | cursor | integer | 必填 | 首次传 0，后续传上次返回的游标 | | MATCH pattern | string | \* | 键名匹配模式（glob风格） | | COUNT count | integer | 10 | 每次扫描的大致数量（非精确） | | TYPE type | string | 无 | 按数据类型过滤（Redis 6.0+） | ```bash # 完整迭代所有键 SCAN 0 # 返回 [新游标, [key1, key2, ...]] # 当返回游标为 0 时迭代完成 SCAN 0 MATCH "user:*" COUNT 100 TYPE hash ``` ### KEYS 返回匹配模式的所有键，生产环境禁止使用（O(n) 阻塞）。 ``` KEYS pattern ``` ### OBJECT FREQ / OBJECT IDLETIME 查看键的访问频率（LFU 策略下）/空闲时间（LRU 策略下）。 ### DUMP / RESTORE 序列化/反序列化键，用于键迁移。 ### COPY 复制键到新键（Redis 6.2+）。 ``` COPY source dest [DB destdb] [REPLACE] ``` ### WAIT 等待指定数量的副本确认写入，用于强一致性场景。 ``` WAIT numreplicas timeout ``` --- ## Python 集成：redis-py 安装： ```bash pip install redis # 异步支持（基于 asyncio）： pip install redis[hiredis] # hiredis C 解析器，性能更好 ``` ### StrictRedis / Redis `redis.StrictRedis` 是主要客户端类（`redis.Redis` 是别名）。 ```python import redis r = redis.StrictRedis( host='localhost', port=6379, db=0, password=None, socket_timeout=None, socket_connect_timeout=None, socket_keepalive=False, socket_keepalive_options=None, connection_pool=None, unix_socket_path=None, encoding='utf-8', encoding_errors='strict', charset=None, errors=None, decode_responses=False, retry_on_timeout=False, retry_on_error=None, ssl=False, ssl_keyfile=None, ssl_certfile=None, ssl_cert_reqs='required', ssl_ca_certs=None, ssl_ca_data=None, ssl_check_hostname=False, ssl_password=None, ssl_validate_ocsp=False, max_connections=None, single_connection_client=False, health_check_interval=0, client_name=None, username=None, retry=None, redis_connect_func=None, credential_provider=None, ) ``` #### StrictRedis 构造参数 | 参数 | 类型 | 默认值 | 说明 | | ------------------------ | ------------------- | ----------- | --------------------- | | host | str | 'localhost' | Redis 服务器地址 | | port | int | 6379 | 端口号 | | db | int | 0 | 数据库编号（0-15） | | password | str/None | None | 认证密码 | | socket\_timeout | float/None | None | 读写超时（秒），None 表示永不超时 | | socket\_connect\_timeout | float/None | None | 连接超时（秒） | | socket\_keepalive | bool | False | 开启 TCP keepalive | | connection\_pool | ConnectionPool/None | None | 复用已有连接池 | | decode\_responses | bool | False | True 时自动将响应字节解码为字符串 | | retry\_on\_timeout | bool | False | 超时时自动重试 | | ssl | bool | False | 启用 TLS/SSL | | max\_connections | int/None | None | 连接池最大连接数（独立连接模式下有效） | | health\_check\_interval | int | 0 | 连接健康检查间隔（秒），0 表示禁用 | | client\_name | str/None | None | 客户端名称（CLIENT SETNAME） | | username | str/None | None | ACL 用户名（Redis 6.0+） | ```python # 常用初始化方式 r = redis.StrictRedis( host='localhost', port=6379, db=0, decode_responses=True, # 推荐：自动解码，避免手动 .decode() socket_timeout=5, socket_connect_timeout=3, ) ``` #### 基本操作 ```python # String r.set('name', 'Alice', ex=3600) # ex=过期秒数，px=过期毫秒数 r.set('lock', '1', nx=True, ex=30) # NX 模式 r.get('name') # b'Alice'（decode_responses=False） r.mset({'a': 1, 'b': 2, 'c': 3}) r.mget(['a', 'b', 'c']) r.incr('counter') r.incrby('counter', 10) r.incrbyfloat('price', -9.99) # List r.rpush('queue', 'task1', 'task2') r.lpop('queue') r.brpop(['queue'], timeout=5) # 阻塞弹出 r.lrange('queue', 0, -1) # Hash r.hset('user:1001', mapping={'name': 'Alice', 'age': 30}) r.hget('user:1001', 'name') r.hmget('user:1001', ['name', 'age']) r.hgetall('user:1001') # 返回 dict r.hincrby('user:1001', 'age', 1) # Set r.sadd('tags', 'python', 'redis') r.smembers('tags') # 返回 set r.sismember('tags', 'python') r.srandmember('tags', 3) r.sunion('tags:a', 'tags:b') r.sinterstore('common', 'friends:1', 'friends:2') # ZSet r.zadd('ranking', {'Alice': 9800, 'Bob': 8700}) r.zadd('ranking', {'Dave': 9000}, nx=True) r.zrange('ranking', 0, -1, withscores=True, rev=True) r.zrank('ranking', 'Alice') r.zincrby('ranking', 100, 'Alice') r.zrangebyscore('ranking', 9000, '+inf', withscores=True) # 通用 r.exists('name') r.delete('name', 'counter') r.expire('session', 1800) r.ttl('session') r.type('ranking') ``` ### ConnectionPool 连接池管理多个连接，避免每次操作都创建新连接。 ```python import redis pool = redis.ConnectionPool( host='localhost', port=6379, db=0, password=None, max_connections=50, decode_responses=True, socket_timeout=5, socket_connect_timeout=3, retry_on_timeout=True, health_check_interval=30, ) r = redis.StrictRedis(connection_pool=pool) ``` #### ConnectionPool 参数 | 参数 | 类型 | 默认值 | 说明 | | ------------------------ | ---------- | ----------- | ----------------------------- | | host | str | 'localhost' | 服务器地址 | | port | int | 6379 | 端口 | | db | int | 0 | 数据库编号 | | max\_connections | int | 无限制 | 连接池最大连接数，超出时抛 ConnectionError | | decode\_responses | bool | False | 自动解码响应 | | socket\_timeout | float/None | None | 读写超时 | | socket\_connect\_timeout | float/None | None | 连接超时 | | retry\_on\_timeout | bool | False | 超时重试 | | health\_check\_interval | int | 0 | 健康检查间隔（秒） | ```python # UnixDomainSocket 连接池（本机性能最佳） unix_pool = redis.ConnectionPool.from_url("unix:///tmp/redis.sock?db=0") # URL 方式初始化 url_pool = redis.ConnectionPool.from_url("redis://:password@localhost:6379/0") r = redis.StrictRedis(connection_pool=url_pool) ``` ### Pipeline Pipeline 将多个命令打包一次性发送，减少网络往返（RTT）。默认非事务性（只打包），指定 `transaction=True` 则等价于 MULTI/EXEC 事务。 ```python # 非事务 Pipeline（仅批量发送） pipe = r.pipeline(transaction=False) pipe.set('a', 1) pipe.set('b', 2) pipe.incr('a') results = pipe.execute() # [True, True, 2] # 事务 Pipeline（MULTI/EXEC） with r.pipeline() as pipe: pipe.multi() # 显式开启事务（pipeline() 默认已是 transaction=True） pipe.set('x', 10) pipe.incr('x') results = pipe.execute() ``` #### pipeline() 方法参数 | 参数 | 类型 | 默认值 | 说明 | | ----------- | -------- | ---- | --------------------------------------- | | transaction | bool | True | True 时使用 MULTI/EXEC 包裹（事务），False 时仅批量发送 | | shard\_hint | str/None | None | 集群模式下指定分片（通常不需要） | #### 带重试的乐观锁（WATCH + Pipeline） ```python def transfer(r, from_key, to_key, amount): with r.pipeline() as pipe: while True: try: pipe.watch(from_key, to_key) # 监视键 balance = int(pipe.get(from_key) or 0) if balance < amount: raise ValueError("余额不足") pipe.multi() # 开启事务 pipe.decrby(from_key, amount) pipe.incrby(to_key, amount) pipe.execute() # 提交，若被监视键有修改则抛 WatchError break except redis.WatchError: continue # 重试 ``` ### 异步客户端（asyncio） ```python import redis.asyncio as aioredis async def main(): r = aioredis.StrictRedis( host='localhost', port=6379, db=0, decode_responses=True, ) await r.set('key', 'value') val = await r.get('key') await r.aclose() # 异步连接池 pool = aioredis.ConnectionPool.from_url("redis://localhost:6379/0") r = aioredis.StrictRedis(connection_pool=pool) ``` ### Lua 脚本（原子复合操作） ```python # eval 执行脚本 script = """ local val = redis.call('GET', KEYS[1]) if val == ARGV[1] then return redis.call('DEL', KEYS[1]) end return 0 """ # 原子删除（仅当值匹配时）—— 分布式锁释放的正确姿势 result = r.eval(script, 1, 'lock_key', 'lock_value') # 预注册脚本（更高效，避免重复传输脚本） sha = r.script_load(script) result = r.evalsha(sha, 1, 'lock_key', 'lock_value') ``` #### eval 参数 | 参数 | 类型 | 默认值 | 说明 | | ----------------- | --- | --- | --------------------------- | | script | str | 必填 | Lua 脚本内容 | | numkeys | int | 必填 | KEYS 数组的长度 | | \*keys\_and\_args | str | 必填 | 先传 key，再传 arg，共 numkeys+N 个 | --- ## 持久化：RDB 与 AOF ### RDB（Redis Database） RDB 在指定时间间隔内，将内存数据快照保存到磁盘 `.rdb` 文件。 **触发方式** | 方式 | 说明 | | ------------- | --------------------------------- | | 自动（配置触发） | save ，满足条件自动触发 | | BGSAVE | 后台异步执行快照（fork 子进程） | | SAVE | 同步执行，阻塞主线程（禁止在生产中使用） | | SHUTDOWN SAVE | 关闭时保存 | | 主从复制 | 主节点生成 RDB 发送给从节点 | **redis.conf 配置** ```ini # 默认配置：900秒内至少1次修改，300秒内至少10次，60秒内至少10000次 save 900 1 save 300 10 save 60 10000 dbfilename dump.rdb dir /var/lib/redis/ # 快照失败时停止写入 stop-writes-on-bgsave-error yes # 使用 LZF 压缩 rdbcompression yes # 校验和 rdbchecksum yes ``` **优缺点** | 优点 | 缺点 | | ----------------- | ------------------- | | 文件紧凑，恢复速度快 | 可能丢失最近一次快照后的数据 | | 对性能影响小（fork 后台执行） | fork 时内存占用翻倍（COW机制） | | 适合灾备、全量备份 | 数据集大时 fork 耗时较长 | ### AOF（Append Only File） AOF 将每个写命令追加到 `.aof` 文件，重启时重放命令恢复数据。 **redis.conf 配置** ```ini appendonly yes appendfilename "appendonly.aof" dir /var/lib/redis/ # 同步策略（核心配置） appendfsync everysec # always / everysec / no # 自动重写（压缩） auto-aof-rewrite-percentage 100 # AOF 文件比上次重写后增长 100% 时触发 auto-aof-rewrite-min-size 64mb # 且文件大小 >= 64mb # 重写期间不同步（减少延迟，但有数据丢失风险） no-appendfsync-on-rewrite no # 允许加载不完整的 AOF（断电场景） aof-use-rdb-preamble yes # AOF 文件以 RDB 格式开头（混合持久化，Redis 4.0+） ``` **appendfsync 选项对比** | 选项 | 说明 | 数据安全性 | 性能 | | -------- | --------------- | --------- | -- | | always | 每次写命令立即 fsync | 最安全，最多丢1条 | 最慢 | | everysec | 每秒 fsync 一次（推荐） | 最多丢1秒数据 | 较好 | | no | 由 OS 决定何时 fsync | 最不安全 | 最快 | **AOF 重写** ```bash BGREWRITEAOF # 手动触发后台重写（合并冗余命令，压缩文件） ``` ### RDB + AOF 混合持久化（推荐，Redis 4.0+） ```ini aof-use-rdb-preamble yes ``` AOF 文件以 RDB 数据开头，后接 AOF 增量命令，兼顾恢复速度和数据安全性。 ### 持久化方案选择 | 场景 | 推荐方案 | | --------- | --------------------- | | 缓存（允许丢失） | 不持久化或仅 RDB | | 一般业务 | AOF everysec + RDB 混合 | | 高可靠（金融） | AOF always 或双写 | | 只做主从复制的从库 | 从库不持久化，主库 RDB | --- ## 发布订阅 Redis 的 Pub/Sub 是简单的消息广播模型：发布者将消息发布到频道，所有订阅该频道的订阅者都会收到。消息不持久化，离线订阅者会错过消息。 ### 基本命令 ```bash # 订阅频道（阻塞等待） SUBSCRIBE channel [channel ...] # 取消订阅 UNSUBSCRIBE [channel ...] # 发布消息（返回接收者数量） PUBLISH channel message # 模式订阅（支持 glob 通配符） PSUBSCRIBE pattern [pattern ...] # 如：PSUBSCRIBE news.* # 取消模式订阅 PUNSUBSCRIBE [pattern ...] # 查看订阅信息 PUBSUB CHANNELS [pattern] # 活跃频道列表 PUBSUB NUMSUB [channel ...] # 各频道订阅者数量 PUBSUB NUMPAT # 模式订阅总数 ``` ### Python 中的发布订阅 ```python import redis import threading r = redis.StrictRedis(decode_responses=True) # 发布者 def publisher(): import time for i in range(5): r.publish('news', f'消息 {i}') time.sleep(1) # 订阅者 def subscriber(): pubsub = r.pubsub() pubsub.subscribe('news') for message in pubsub.listen(): if message['type'] == 'message': print(f"收到: {message['data']}") # pubsub.subscribe() 参数 # subscribe(channel, **kwargs)：也可传入回调函数 # pubsub.subscribe(**{'news': handler}) # 带回调 # 模式订阅 pubsub = r.pubsub() pubsub.psubscribe(**{'news.*': lambda msg: print(msg['data'])}) pubsub.run_in_thread(sleep_time=0.01, daemon=True) # 后台线程处理 ``` #### pubsub 方法参数 | 方法 | 参数 | 说明 | | --------------------------------------------------------------------- | ------------- | ------------------------- | | subscribe(\*channels) | channels: str | 订阅频道，可传回调：{'ch': handler} | | unsubscribe(\*channels) | channels: str | 取消订阅，不传则取消所有 | | psubscribe(\*patterns) | patterns: str | 模式订阅 | | punsubscribe(\*patterns) | patterns: str | 取消模式订阅 | | listen() | 无 | 生成器，阻塞迭代消息 | | get\_message(ignore\_subscribe\_messages=False, timeout=0) | 见下 | 非阻塞获取消息 | | run\_in\_thread(sleep\_time=0, daemon=False, exception\_handler=None) | 见下 | 启动后台线程处理消息 | #### get\_message 参数 | 参数 | 类型 | 默认值 | 说明 | | --------------------------- | ----- | ----- | -------------- | | ignore\_subscribe\_messages | bool | False | 忽略订阅/取消订阅的确认消息 | | timeout | float | 0 | 等待消息的超时秒数 | #### run\_in\_thread 参数 | 参数 | 类型 | 默认值 | 说明 | | ------------------ | ------------- | ----- | ----------------------------- | | sleep\_time | float | 0 | 每次轮询的间隔（秒） | | daemon | bool | False | 是否设为守护线程 | | exception\_handler | callable/None | None | 异常处理函数，签名：(e, pubsub, thread) | ### 消息结构 ```python { 'type': 'message', # message / pmessage / subscribe / unsubscribe 等 'pattern': None, # 模式订阅时为匹配的模式字符串 'channel': 'news', # 频道名 'data': '消息内容' # 消息内容（subscribe 类型时为订阅数量） } ``` ### Pub/Sub 的局限与替代 | 问题 | 说明 | 替代方案 | | ----- | ---------------- | ------------------------- | | 不持久化 | 订阅者离线期间的消息丢失 | Redis Streams（XADD/XREAD） | | 无 ACK | 消息发出不确认是否被处理 | Redis Streams / 专业 MQ | | 无消费组 | 同一消息所有订阅者都收到（广播） | Redis Streams XGROUP | --- ## 过期策略与内存淘汰 ### 键过期策略 Redis 使用两种机制配合删除过期键： | 机制 | 说明 | | ------------ | ------------------------------- | | 惰性删除（Lazy） | 访问键时检查是否过期，过期则删除。节省 CPU，但可能浪费内存 | | 定期删除（Active） | 每 100ms 随机采样一批设有过期时间的键，删除其中过期的键 | 定期删除算法：每次随机取 20 个带 TTL 的键，若过期比例超过 25%，则重复采样，直到比例降低或达到时间上限（默认 25ms）。 ### 内存淘汰策略 当内存达到 `maxmemory` 限制时，Redis 根据 `maxmemory-policy` 决定如何处理新写入。 **redis.conf 配置** ```ini maxmemory 2gb maxmemory-policy allkeys-lru maxmemory-samples 5 # LRU/LFU 近似算法的采样数量，越大越精确但越慢 ``` **淘汰策略对比** | 策略 | 说明 | 适用场景 | | --------------- | --------------------------- | --------- | | noeviction | 不淘汰，写入超出限制时报错 | 不能丢数据 | | allkeys-lru | 对所有键使用 LRU 算法淘汰 | 通用缓存 | | allkeys-lfu | 对所有键使用 LFU 算法淘汰（Redis 4.0+） | 热点数据明显时 | | allkeys-random | 随机淘汰所有键 | 均匀访问 | | volatile-lru | 对有 TTL 的键使用 LRU 淘汰 | 持久数据+缓存混用 | | volatile-lfu | 对有 TTL 的键使用 LFU 淘汰 | 同上 | | volatile-random | 随机淘汰有 TTL 的键 | 同上 | | volatile-ttl | 淘汰 TTL 最短的键 | 短期缓存优先淘汰 | **LRU vs LFU** - LRU（最近最少使用）：淘汰最久没有访问的键，适合大多数场景。 - LFU（最不常用）：淘汰访问次数最少的键，对热点数据更友好，避免冷数据一次访问后不被淘汰。 --- ## 分布式锁 ### 基础实现 利用 `SET key value NX EX` 实现分布式锁： ```python import uuid import redis import time r = redis.StrictRedis(decode_responses=True) def acquire_lock(lock_name: str, expire: int = 30) -> str | None: """ 获取分布式锁。 返回锁标识（用于释放），获取失败返回 None。 """ identifier = str(uuid.uuid4()) acquired = r.set(lock_name, identifier, nx=True, ex=expire) return identifier if acquired else None def release_lock(lock_name: str, identifier: str) -> bool: """ 释放分布式锁（Lua 脚本保证原子性）。 """ script = """ if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end """ result = r.eval(script, 1, lock_name, identifier) return bool(result) # 使用示例 def critical_section(): lock_name = 'lock:order:create' identifier = acquire_lock(lock_name, expire=30) if not identifier: raise Exception("获取锁失败，稍后重试") try: # 执行需要互斥的业务逻辑 pass finally: release_lock(lock_name, identifier) ``` ### 上下文管理器封装 ```python import contextlib import uuid import time import redis class RedisLock: def __init__( self, client: redis.StrictRedis, name: str, expire: int = 30, retry_times: int = 3, retry_delay: float = 0.1, ): """ 参数说明： client - redis.StrictRedis 实例 name - 锁的键名 expire - 锁过期时间（秒），防止死锁 retry_times - 获取锁失败时的重试次数 retry_delay - 每次重试的间隔（秒） """ self.client = client self.name = name self.expire = expire self.retry_times = retry_times self.retry_delay = retry_delay self.identifier = None _release_script = """ if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end """ def acquire(self) -> bool: self.identifier = str(uuid.uuid4()) for _ in range(self.retry_times): if self.client.set(self.name, self.identifier, nx=True, ex=self.expire): return True time.sleep(self.retry_delay) return False def release(self) -> bool: if self.identifier is None: return False result = self.client.eval(self._release_script, 1, self.name, self.identifier) self.identifier = None return bool(result) def __enter__(self): if not self.acquire(): raise TimeoutError(f"获取锁 {self.name!r} 失败") return self def __exit__(self, *args): self.release() # 使用 with RedisLock(r, 'lock:order:create', expire=30): # 临界区 pass ``` ### Redlock 算法（多节点高可用锁） 单节点锁在 Redis 主从切换时可能失效（主宕机，从未同步锁）。Redlock 在多个独立 Redis 节点上加锁，多数节点成功才算获锁成功。 ```bash pip install redlock-py ``` ```python from redlock import Redlock, MultipleRedlockException dlm = Redlock([ {"host": "localhost", "port": 6379, "db": 0}, {"host": "localhost", "port": 6380, "db": 0}, {"host": "localhost", "port": 6381, "db": 0}, ]) try: lock = dlm.lock("my_resource", 10000) # 10000ms 过期 if lock: try: # 临界区 pass finally: dlm.unlock(lock) except MultipleRedlockException: print("获取锁失败") ``` Redlock 要求至少 3 个（推荐 5 个）独立 Redis 实例，且各节点时钟误差应尽量小。 --- ## 最佳实践 ### 键设计 ```python # 好的键名：有层级、有业务含义、长度适中 "user:profile:1001" "cache:api:v2:product_list:page_1" "session:token:abc123def456" # 避免： "u1001" # 无可读性 "user_profile_1001_cache_data_json" # 太长，无层级 ``` ### 连接管理 ```python # 应用启动时创建单一全局连接池 import redis _pool = redis.ConnectionPool( host='redis-host', port=6379, db=0, max_connections=20, decode_responses=True, socket_timeout=3, socket_connect_timeout=2, retry_on_timeout=True, health_check_interval=30, ) def get_redis() -> redis.StrictRedis: return redis.StrictRedis(connection_pool=_pool) ``` ### 批量操作用 Pipeline ```python # 差：N 次网络往返 for user_id in user_ids: r.hset(f'user:{user_id}', 'active', 1) # 好：1 次网络往返 with r.pipeline(transaction=False) as pipe: for user_id in user_ids: pipe.hset(f'user:{user_id}', 'active', 1) pipe.execute() ``` ### 大数据集使用 SCAN 替代 KEYS ```python def scan_all_keys(r, pattern='*', count=100): """安全迭代所有匹配键，不阻塞服务器""" cursor = 0 while True: cursor, keys = r.scan(cursor, match=pattern, count=count) for key in keys: yield key if cursor == 0: break # 使用 for key in scan_all_keys(r, pattern='user:*', count=200): process(key) ``` ### 设置合理的过期时间 ```python # 避免大量键同时过期（缓存雪崩）：加随机偏移 import random base_ttl = 3600 ttl = base_ttl + random.randint(-300, 300) # ±5分钟随机偏移 r.set('cache:product_list', data, ex=ttl) ``` ### 缓存穿透防护 ```python def get_product(product_id: int): cache_key = f'product:{product_id}' data = r.get(cache_key) if data is None: # 查数据库 product = db.query(product_id) if product is None: # 缓存空值，防止穿透（给较短过期时间） r.set(cache_key, '', ex=60) return None r.set(cache_key, serialize(product), ex=3600) return product elif data == '': # 已知不存在 return None else: return deserialize(data) ``` ### 热点键（大 Key）处理 ```python # 检测大 Key（命令行） redis-cli --bigkeys # 对于大 Hash，按字段前缀分拆 # 代替 HSET user:1001 field1 v1 field2 v2 ... (10000字段) # 改为： HSET user:1001:basic name age email HSET user:1001:stats login_count last_seen HSET user:1001:prefs theme language ``` --- ## 常见陷阱与注意事项 ### 1\. decode\_responses 未启用导致字节类型问题 ```python r = redis.StrictRedis() # 默认 decode_responses=False val = r.get('name') # b'Alice'（字节，非字符串） val.upper() # 正常，但比较时容易出错 # 推荐：始终启用 r = redis.StrictRedis(decode_responses=True) val = r.get('name') # 'Alice'（字符串） ``` ### 2\. Pipeline 中异常处理 ```python # Pipeline 默认遇到错误会继续执行，返回结果列表中对应位置为 ResponseError pipe = r.pipeline(transaction=False) pipe.set('a', 1) pipe.lpush('a', 'x') # 错误：a 是 string 类型，不能 lpush pipe.set('b', 2) results = pipe.execute(raise_on_error=False) # [True, ResponseError, True] # raise_on_error=True（默认）时，遇到错误立即抛出 ``` ### 3\. KEYS 命令阻塞生产环境 ```python # 禁止在生产中使用： r.keys('user:*') # O(n)，阻塞所有其他命令 # 正确做法： for key in scan_all_keys(r, 'user:*'): pass ``` ### 4\. 分布式锁忘记设过期时间导致死锁 ```python # 错误：不设 EX，进程崩溃后锁永远不释放 r.set('lock', '1', nx=True) # 正确：必须同时设 NX 和 EX（原子操作，不能分两步） r.set('lock', identifier, nx=True, ex=30) ``` ### 5\. 锁释放时未校验持有者 ```python # 错误：直接 DEL，可能删除别人的锁（自己的锁超时后别人加锁，自己又去删） r.delete('lock') # 正确：用 Lua 脚本原子校验并删除 script = "if redis.call('GET',KEYS[1])==ARGV[1] then return redis.call('DEL',KEYS[1]) else return 0 end" r.eval(script, 1, 'lock', my_identifier) ``` ### 6\. ZSet score 精度问题 ```python # score 为 double（64位浮点），大整数可能丢失精度 ZADD ranking 9007199254740993 "user" # 超过 2^53，精度丢失 # 解决：用时间戳的毫秒值（在安全整数范围内即可） # 或者将精度要求高的 score 数据存到 Hash，ZSet 只做排序索引 ``` ### 7\. EXPIRE 对已有键的覆盖行为 ```python r.set('key', 'value', ex=60) r.set('key', 'new_value') # 不带 ex，覆盖后 TTL 消失！变为永久有效 # 正确：更新值时保持 TTL ttl = r.ttl('key') r.set('key', 'new_value', ex=ttl if ttl > 0 else 60) # 或使用 KEEPTTL 选项（Redis 6.0+） r.set('key', 'new_value', keepttl=True) ``` ### 8\. 哈希存储 Python 类型时自动序列化 ```python r = redis.StrictRedis(decode_responses=True) r.hset('data', 'count', 100) # 存为字符串 "100" r.hget('data', 'count') # "100"（字符串，非整数） int(r.hget('data', 'count')) + 1 # 需要手动转换 r.hset('data', 'flag', True) # 存为 "True"（字符串！） r.hget('data', 'flag') == True # False！ # 建议：布尔/复杂类型用 JSON 序列化，或用 HINCRBY 处理数值 ``` ### 9\. Pub/Sub 消息丢失 ```python # Pub/Sub 无持久化，订阅者断线期间的消息全部丢失 # 需要可靠消息传递时，使用 Redis Streams： r.xadd('stream:events', {'event': 'user_login', 'uid': '1001'}) r.xread({'stream:events': '$'}, count=10, block=1000) ``` ### 10\. 连接池耗尽 ```python # 症状：ConnectionError: Too many connections # 原因：max_connections 太小，或连接未释放（如 Pipeline 未 execute） # 检查连接数 r.execute_command('CLIENT', 'LIST') # 解决： # 1. 适当增大 max_connections # 2. 始终用 with 语句使用 Pipeline # 3. 检查是否有连接泄漏（未关闭的 pubsub 等） ``` ### 11\. 缓存雪崩、穿透、击穿 | 问题 | 原因 | 解决方案 | | -- | ---------------- | ---------------------- | | 雪崩 | 大量缓存同时过期 | TTL 加随机偏移；多级缓存；限流降级 | | 穿透 | 查询不存在的键，每次都打到数据库 | 缓存空值；布隆过滤器 | | 击穿 | 热点键过期瞬间大量请求涌入 | 分布式锁串行重建；不设过期时间；提前异步刷新 | ### 12\. 事务（MULTI/EXEC）不支持回滚 Redis 事务中命令执行错误（如类型错误）不会回滚已执行的命令，只有入队时的语法错误才会中止整个事务。Redis 事务只保证隔离性（事务中命令不被打断），不保证原子性（部分失败不回滚）。 ```python with r.pipeline() as pipe: pipe.set('a', 'hello') pipe.incr('a') # 运行时错误（a 不是整数），不会导致 set('a') 回滚 results = pipe.execute(raise_on_error=False) # results[0] = True（set 成功） # results[1] = ResponseError（incr 失败） # a 的值仍为 'hello'，set 未回滚 ``` --- ## 参见 [装饰器与函数高级](https://blog.vercanti.com/python-zhuang-shi-qi-yu-han-shu-gao-ji-yong-fa/) [MySQL基础完全指南](https://blog.vercanti.com/mysql-ji-chu-wan-quan-zhi-nan/) [ClickHouse完全指南](https://blog.vercanti.com/clickhouse-wan-quan-zhi-nan/)