Redis 完全指南

最后更新：2026-03-05

官方文档：https://redis.io/docs/
redis-py 文档：https://redis-py.readthedocs.io/

目录

1. Redis完全指南 · 基础概念
2. Redis完全指南 · String（字符串）
3. Redis完全指南 · List（列表）
4. Redis完全指南 · Hash（哈希）
5. Redis完全指南 · Set（集合）
6. Redis完全指南 · ZSet（有序集合）
7. Redis完全指南 · 通用键命令
8. Redis完全指南 · Python 集成：redis-py
9. Redis完全指南 · 持久化：RDB 与 AOF
10. Redis完全指南 · 发布订阅
11. Redis完全指南 · 过期策略与内存淘汰
12. Redis完全指南 · 分布式锁
13. Redis完全指南 · 最佳实践
14. Redis完全指南 · 常见陷阱与注意事项

基础概念

Redis（Remote Dictionary Server）是一个开源的、基于内存的数据结构存储系统，可用作数据库、缓存和消息代理。

核心特性

键命名规范

业务:对象: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]

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

GET name      # "Alice"
GET noexist   # (nil)

MSET / MGET

批量设置/获取，减少网络往返。

MSET key value [key value ...]
MGET key [key ...]

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

INCR page_view           # 1（键不存在从0开始）
INCRBY page_view 10      # 11
INCRBYFLOAT price 0.5    # 浮点递增
DECR stock               # 库存递减

APPEND

追加值到已有字符串末尾。

APPEND key value

APPEND log "2024-01-01 login\n"
STRLEN log    # 返回字节长度

GETRANGE / SETRANGE

操作字符串的子串。

GETRANGE key start end
SETRANGE key offset value

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 ...]

RPUSH queue "task1" "task2" "task3"   # 右进：[task1, task2, task3]
LPUSH stack "a" "b" "c"              # 左进：[c, b, a]

LPOP / RPOP

从左端/右端弹出元素。

LPOP key [count]
RPOP key [count]

LPOP queue          # "task1"（消费队列头部）
RPOP stack          # "a"（弹出栈顶）
LPOP queue 2        # 一次弹出2个，返回列表

BLPOP / BRPOP

阻塞版弹出，队列为空时阻塞等待，适合实现消费者模式。

BLPOP key [key ...] timeout
BRPOP key [key ...] timeout

BLPOP queue 5      # 阻塞最多5秒等待消息，返回 [键名, 值]
BLPOP q1 q2 0      # 永久阻塞，监听多个队列

LRANGE

获取列表片段（不修改列表）。

LRANGE key start stop

LRANGE queue 0 -1    # 获取全部
LRANGE queue 0 9     # 获取前10个
LRANGE queue -5 -1   # 获取最后5个

LLEN

返回列表长度。

LLEN key

LLEN queue    # 队列中的元素数量

LINDEX

按索引获取元素，O(n) 复杂度，慎用于大列表。

LINDEX key index

LSET

按索引设置元素值。

LSET key index element

LINSERT

在指定元素前/后插入新元素，O(n) 复杂度。

LINSERT key BEFORE|AFTER pivot element

LREM

删除列表中匹配的元素。

LREM key count element

LTRIM

裁剪列表，只保留指定范围。

LTRIM key start stop

# 只保留最新的100条
RPUSH logs "entry"
LTRIM logs -100 -1

Hash（哈希）

Hash 是一个键值对集合，适合存储对象。底层使用压缩列表（listpack）或哈希表。

适用场景：存储用户信息、商品详情等结构化对象，比 JSON 字符串更灵活（可单字段更新）。

HSET

设置一个或多个字段（Redis 4.0+ 支持多字段）。

HSET key field value [field value ...]

HSET user:1001 name "Alice" age 30 email "[email protected]"
HSET user:1001 age 31       # 更新单个字段

HGET / HMGET

获取一个或多个字段值。

HGET key field
HMGET key field [field ...]

HGET user:1001 name         # "Alice"
HMGET user:1001 name age    # ["Alice", "31"]

HGETALL

获取所有字段和值，返回扁平化列表（字段名/值交替）。

HGETALL key

HGETALL user:1001
# 1) "name"
# 2) "Alice"
# 3) "age"
# 4) "31"
# 5) "email"
# 6) "[email protected]"

注意：大哈希（字段数量多）调用 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

HINCRBY user:1001 login_count 1
HINCRBYFLOAT user:1001 balance -9.99

HSCAN

游标式迭代哈希字段，避免阻塞。

HSCAN key cursor [MATCH pattern] [COUNT count]

HSCAN user:1001 0 MATCH "addr*" COUNT 100

Set（集合）

Set 是无序、不重复的字符串集合。底层使用 listpack 或哈希表。

适用场景：标签系统、好友关系、抽奖（去重随机选取）、UV 计数。

SADD

添加一个或多个元素。

SADD key member [member ...]

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 ...]

SISMEMBER tags "python"            # 1
SMISMEMBER tags "python" "java"    # [1, 0]

SCARD

返回集合元素数量（基数）。

SCARD key

SRANDMEMBER

随机返回集合中的元素（不删除）。

SRANDMEMBER key [count]

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

# 共同好友
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 ...]

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

# 获取排行榜前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]

ZRANK leaderboard "Alice"      # 从低到高的排名
ZREVRANK leaderboard "Alice"   # 从高到低的排名（第1名返回0）

ZSCORE

获取元素的分数。

ZSCORE key member

ZINCRBY

对元素的分数原子递增。

ZINCRBY key increment member

ZINCRBY leaderboard 100 "Alice"   # Alice 加分

ZREM

删除一个或多个元素。

ZREM key member [member ...]

ZCARD

返回有序集合元素总数。

ZCARD key

ZCOUNT

返回分数在指定范围内的元素数量。

ZCOUNT key min max

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（不含）、-（最小）、+（最大）。

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 ...]

# 综合排行：两个榜单加权求和
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]

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

SCAN

游标式迭代所有键，替代 KEYS（KEYS 会阻塞服务器）。

SCAN cursor [MATCH pattern] [COUNT count] [TYPE type]

# 完整迭代所有键
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

安装：

pip install redis
# 异步支持（基于 asyncio）：
pip install redis[hiredis]   # hiredis C 解析器，性能更好

StrictRedis / Redis

redis.StrictRedis 是主要客户端类（redis.Redis 是别名）。

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 构造参数

# 常用初始化方式
r = redis.StrictRedis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,  # 推荐：自动解码，避免手动 .decode()
    socket_timeout=5,
    socket_connect_timeout=3,
)

基本操作

# 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

连接池管理多个连接，避免每次操作都创建新连接。

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 参数

# 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 事务。

# 非事务 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() 方法参数

带重试的乐观锁（WATCH + Pipeline）

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）

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 脚本（原子复合操作）

# 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 参数

持久化：RDB 与 AOF

RDB（Redis Database）

RDB 在指定时间间隔内，将内存数据快照保存到磁盘 .rdb 文件。

触发方式

redis.conf 配置

# 默认配置：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

优缺点

AOF（Append Only File）

AOF 将每个写命令追加到 .aof 文件，重启时重放命令恢复数据。

redis.conf 配置

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 选项对比

AOF 重写

BGREWRITEAOF    # 手动触发后台重写（合并冗余命令，压缩文件）

RDB + AOF 混合持久化（推荐，Redis 4.0+）

aof-use-rdb-preamble yes

AOF 文件以 RDB 数据开头，后接 AOF 增量命令，兼顾恢复速度和数据安全性。

持久化方案选择

发布订阅

Redis 的 Pub/Sub 是简单的消息广播模型：发布者将消息发布到频道，所有订阅该频道的订阅者都会收到。消息不持久化，离线订阅者会错过消息。

基本命令

# 订阅频道（阻塞等待）
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 中的发布订阅

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 方法参数

get_message 参数

run_in_thread 参数

消息结构

{
    'type': 'message',        # message / pmessage / subscribe / unsubscribe 等
    'pattern': None,          # 模式订阅时为匹配的模式字符串
    'channel': 'news',        # 频道名
    'data': '消息内容'        # 消息内容（subscribe 类型时为订阅数量）
}

Pub/Sub 的局限与替代

过期策略与内存淘汰

键过期策略

Redis 使用两种机制配合删除过期键：

定期删除算法：每次随机取 20 个带 TTL 的键，若过期比例超过 25%，则重复采样，直到比例降低或达到时间上限（默认 25ms）。

内存淘汰策略

当内存达到 maxmemory 限制时，Redis 根据 maxmemory-policy 决定如何处理新写入。

redis.conf 配置

maxmemory 2gb
maxmemory-policy allkeys-lru
maxmemory-samples 5     # LRU/LFU 近似算法的采样数量，越大越精确但越慢

淘汰策略对比

LRU vs LFU

• LRU（最近最少使用）：淘汰最久没有访问的键，适合大多数场景。
• LFU（最不常用）：淘汰访问次数最少的键，对热点数据更友好，避免冷数据一次访问后不被淘汰。

分布式锁

基础实现

利用 SET key value NX EX 实现分布式锁：

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)

上下文管理器封装

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 节点上加锁，多数节点成功才算获锁成功。

pip install redlock-py

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 实例，且各节点时钟误差应尽量小。

最佳实践

键设计

# 好的键名：有层级、有业务含义、长度适中
"user:profile:1001"
"cache:api:v2:product_list:page_1"
"session:token:abc123def456"

# 避免：
"u1001"              # 无可读性
"user_profile_1001_cache_data_json"   # 太长，无层级

连接管理

# 应用启动时创建单一全局连接池
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

# 差：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

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)

设置合理的过期时间

# 避免大量键同时过期（缓存雪崩）：加随机偏移
import random

base_ttl = 3600
ttl = base_ttl + random.randint(-300, 300)   # ±5分钟随机偏移
r.set('cache:product_list', data, ex=ttl)

缓存穿透防护

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）处理

# 检测大 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 未启用导致字节类型问题

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 中异常处理

# 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 命令阻塞生产环境

# 禁止在生产中使用：
r.keys('user:*')   # O(n)，阻塞所有其他命令

# 正确做法：
for key in scan_all_keys(r, 'user:*'):
    pass

4. 分布式锁忘记设过期时间导致死锁

# 错误：不设 EX，进程崩溃后锁永远不释放
r.set('lock', '1', nx=True)

# 正确：必须同时设 NX 和 EX（原子操作，不能分两步）
r.set('lock', identifier, nx=True, ex=30)

5. 锁释放时未校验持有者

# 错误：直接 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 精度问题

# score 为 double（64位浮点），大整数可能丢失精度
ZADD ranking 9007199254740993 "user"   # 超过 2^53，精度丢失

# 解决：用时间戳的毫秒值（在安全整数范围内即可）
# 或者将精度要求高的 score 数据存到 Hash，ZSet 只做排序索引

7. EXPIRE 对已有键的覆盖行为

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 类型时自动序列化

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 消息丢失

# Pub/Sub 无持久化，订阅者断线期间的消息全部丢失
# 需要可靠消息传递时，使用 Redis Streams：
r.xadd('stream:events', {'event': 'user_login', 'uid': '1001'})
r.xread({'stream:events': '$'}, count=10, block=1000)

10. 连接池耗尽

# 症状：ConnectionError: Too many connections
# 原因：max_connections 太小，或连接未释放（如 Pipeline 未 execute）

# 检查连接数
r.execute_command('CLIENT', 'LIST')

# 解决：
# 1. 适当增大 max_connections
# 2. 始终用 with 语句使用 Pipeline
# 3. 检查是否有连接泄漏（未关闭的 pubsub 等）

11. 缓存雪崩、穿透、击穿

12. 事务（MULTI/EXEC）不支持回滚

Redis 事务中命令执行错误（如类型错误）不会回滚已执行的命令，只有入队时的语法错误才会中止整个事务。Redis 事务只保证隔离性（事务中命令不被打断），不保证原子性（部分失败不回滚）。

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 未回滚

参见

装饰器与函数高级
MySQL基础完全指南
ClickHouse完全指南