本篇由AI生成

Redis

什么是 Redis

Redis（Remote Dictionary Server）是一个基于内存的键值对数据库，性能极高（读写 10万+ 次/秒），常被用来做缓存、会话存储、消息队列等。

Redis 能做什么

场景 说明 示例
缓存 缓存热点数据，降低数据库压力 用户信息、文章列表缓存
Session 存储 替代内存/数据库存储会话数据 Day5 中的 Redis Session
消息队列 用 List 实现简单的生产者-消费者模式 异步任务、延迟队列
排行榜 利用 Sorted Set 排序 积分排行、热搜榜单
计数器 INCR 原子自增 访问量、点赞数
分布式锁 多实例互斥访问共享资源 防止重复下单
发布订阅 Pub/Sub 模式广播消息 实时通知、聊天

与传统数据库的对比

对比项 Redis 关系型数据库（如 MySQL）
数据存储 内存（可持久化到磁盘） 磁盘
读写速度 纳秒级 毫秒级
数据结构 字符串、哈希、列表等 5+ 种 表、行、列
查询方式 Key-Value SQL 查询
数据量 受限于内存大小 可 TB 级
持久化 RDB 快照 / AOF 日志 默认持久化
事务 简单（不支持回滚） ACID 完整事务

安装与连接

Windows 安装

Windows 上推荐使用 Redis for Windows（WSL 也行）：

1
2
3
4
5
# 使用 winget 安装（Windows 10+）
winget install Redis

# 或者用 Docker
docker run -d --name redis -p 6379:6379 redis:7-alpine

Redis 默认端口 6379，无密码。

安装 Python 客户端

1
pip install redis

同步 vs 异步连接

Redis 提供了两套 API：同步（redis.Redis）和异步（redis.asyncio.Redis），在 FastAPI 中必须用异步版，否则会阻塞事件循环。

1
2
3
4
5
6
7
# ❌ 错误：同步连接会阻塞 FastAPI 异步路由
import redis
r = redis.Redis(host="localhost", port=6379)

# ✅ 正确：使用异步连接
import redis.asyncio as aioredis
r = aioredis.Redis(host="localhost", port=6379, decode_responses=True)

连接配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from redis.asyncio import Redis

# 基础连接
redis_client = Redis(
    host="localhost",
    port=6379,
    db=0,                       # 数据库编号（0-15）
    decode_responses=True,      # 自动将 bytes 解码为 str
)

# 带密码的连接
redis_client = Redis(
    host="localhost",
    port=6379,
    password="your_password",
    db=0,
    decode_responses=True,
)

# 连接池（FastAPI 推荐）
redis_client = Redis(
    host="localhost",
    port=6379,
    max_connections=20,         # 连接池最大连接数
    decode_responses=True,
)

使用连接 URL

1
2
3
4
5
6
7
8
9
10
11
12
# 无密码
redis_client = Redis.from_url("redis://localhost:6379/0")

# 有密码
redis_client = Redis.from_url("redis://:password@localhost:6379/0")

# 带连接池配置
redis_client = Redis.from_url(
    "redis://localhost:6379/0",
    max_connections=20,
    decode_responses=True,
)

在 FastAPI 中管理 Redis 生命周期

使用 FastAPI 的生命周期事件初始化/关闭连接：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from contextlib import asynccontextmanager
from fastapi import FastAPI
from redis.asyncio import Redis

redis_client: Redis | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：创建连接
    global redis_client
    redis_client = Redis.from_url(
        "redis://localhost:6379/0",
        max_connections=20,
        decode_responses=True,
    )
    print("✅ Redis 已连接")

    yield

    # 关闭时：关闭连接
    await redis_client.aclose()
    print("🔌 Redis 已断开")

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    await redis_client.set("hello", "world")   # 使用全局 redis_client
    return {"message": "ok"}

FastAPI 的 lifespan 替代了废弃的 on_event("startup") / on_event("shutdown") 写法。

核心数据结构

Redis 有 5 种核心数据结构，每种都是面试高频内容。

1. String（字符串）

最基础的类型，value 最大 512MB。可用于缓存 JSON、计数器、验证码等。

常用命令：

命令 说明 示例
SET key value 设置键值对 SET username "豆棍度子"
GET key 获取值 GET username
SETEX key seconds value 设置 + 过期时间（秒） SETEX captcha 300 1234
SETNX key value 不存在时才设置（分布式锁） SETNX lock:order:1 "locked"
INCR key 原子自增 1 INCR page_view
INCRBY key n 原子增加 n INCRBY page_view 5
DECR key 原子自减 1 DECR stock:1001
DEL key 删除键 DEL username
EXPIRE key seconds 设置过期时间（已存在的键） EXPIRE captcha 300
TTL key 查看剩余存活秒数（-1 永不过期，-2 已过期） TTL captcha
MSET key value [key value ...] 批量设置 MSET a 1 b 2 c 3
MGET key [key ...] 批量获取 MGET a b c

Python 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 基本操作
await redis_client.set("username", "豆棍度子")
username = await redis_client.get("username")     # "豆棍度子"

# 带过期时间
await redis_client.setex("captcha", 300, "1234")  # 5 分钟过期

# 计数器
await redis_client.incr("page_view")              # 1
await redis_client.incrby("page_view", 5)         # +5
await redis_client.decr("stock:1001")             # -1

# 检查是否还有剩余时间（秒）
ttl = await redis_client.ttl("captcha")           # 还剩多少秒

# 批量操作
await redis_client.mset({"a": 1, "b": 2, "c": 3})
values = await redis_client.mget("a", "b", "c")   # [1, 2, 3]

# 删除
await redis_client.delete("captcha")

2. Hash（哈希）

Hash 是一个键值对集合，适合存储对象（如用户信息、商品信息）。

可以把 Hash 想象成 Python 的嵌套字典：redis_key → {field: value, field: value}

常用命令：

命令 说明 示例
HSET key field value 设置 Hash 中的字段 HSET user:1001 username "豆棍度子"
HGET key field 获取 Hash 中某字段 HGET user:1001 username
HGETALL key 获取 Hash 所有字段和值 HGETALL user:1001
HMGET key field [field ...] 批量获取字段 HMGET user:1001 username email
HDEL key field [field ...] 删除字段 HDEL user:1001 password
HEXISTS key field 判断字段是否存在 HEXISTS user:1001 email
HKEYS key 获取所有字段名 HKEYS user:1001
HVALS key 获取所有字段值 HVALS user:1001
HLEN key 获取字段数量 HLEN user:1001
HINCRBY key field n 字段原子增减 HINCRBY user:1001 age 1

Python 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 存储用户信息（对象）
await redis_client.hset("user:1001", mapping={
    "username": "豆棍度子",
    "email": "dgdz@example.com",
    "age": 18,
    "role": "admin",
})

# 获取单个字段
name = await redis_client.hget("user:1001", "username")

# 获取多个字段
name, email = await redis_client.hmget("user:1001", "username", "email")

# 获取所有字段和值
user = await redis_client.hgetall("user:1001")
# {"username": "豆棍度子", "email": "dgdz@example.com", "age": "18", "role": "admin"}

# 判断字段是否存在
has_email = await redis_client.hexists("user:1001", "email")   # True

# 只获取字段名
fields = await redis_client.hkeys("user:1001")

# 只获取值
values = await redis_client.hvals("user:1001")

# 更新某个字段
await redis_client.hset("user:1001", "age", 19)    # 18 → 19
await redis_client.hincrby("user:1001", "age", 1)  # 19 → 20

# 删除某个字段
await redis_client.hdel("user:1001", "role")

什么时候用 Hash 而不是 String 存 JSON？ 当你只需要读写对象的某个字段时，Hash 不用把整个 JSON 取出来再改。

3. List（列表）

Redis 的 List 是双向链表（不是数组），头尾操作性能极高（O(1)），中间操作很慢（O(n)）。适合做消息队列、最新消息列表。

常用命令：

命令 说明 示例
LPUSH key value [value ...] 从左侧推入（insert 头部） LPUSH queue task1
RPUSH key value [value ...] 从右侧推入（insert 尾部） RPUSH queue task1
LPOP key 从左侧弹出（移除并返回） LPOP queue
RPOP key 从右侧弹出 RPOP queue
BLPOP key timeout 阻塞式左侧弹出（没数据就等） BLPOP queue 0
BRPOP key timeout 阻塞式右侧弹出 BRPOP queue 5
LLEN key 列表长度 LLEN queue
LRANGE key start stop 范围获取 LRANGE queue 0 -1（全部）
LTRIM key start stop 裁剪列表（保留区间内） LTRIM queue 0 99（保留前100条）

Python 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 生产者：向队列尾部添加任务
await redis_client.rpush("task_queue", "task1", "task2", "task3")

# 消费者：从队列头部取出任务
task = await redis_client.lpop("task_queue")   # "task1"

# 分页获取列表内容（类似 Python 切片）
tasks = await redis_client.lrange("task_queue", 0, -1)    # 全部
tasks = await redis_client.lrange("task_queue", 0, 9)     # 前 10 条

# 获取列表长度
count = await redis_client.llen("task_queue")

# 阻塞式消费（没有任务就一直等）
task = await redis_client.blpop("task_queue", timeout=0)  # 0=无限等待
# 返回如：("task_queue", "task2")

使用 List 做消息队列：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import asyncio

# 生产者路由
@app.post("/send-task")
async def send_task(task_name: str):
    await redis_client.rpush("task_queue", task_name)
    return {"message": f"任务已提交：{task_name}"}

# 消费者：后台循环消费
async def worker_loop():
    while True:
        result = await redis_client.blpop("task_queue", timeout=0)
        # blpop 返回 ("task_queue", "task_name")
        task_name = result[1]
        print(f"处理任务：{task_name}")
        await asyncio.sleep(1)         # 模拟处理耗时
        print(f"任务完成：{task_name}")

@app.on_event("startup")
async def start_worker():
    asyncio.create_task(worker_loop())  # 启动后台消费者

4. Set（集合）

Set 是无序且元素唯一的集合，基于哈希表实现。适合做去重、共同好友、标签系统。

常用命令：

命令 说明 示例
SADD key member [member ...] 添加元素 SADD tags "后端" "Python"
SREM key member [member ...] 移除元素 SREM tags "前端"
SMEMBERS key 获取所有元素 SMEMBERS tags
SISMEMBER key member 判断是否存在于集合 SISMEMBER tags "Python"
SCARD key 获取元素数量（基数） SCARD tags
SINTER key [key ...] 交集 SINTER set1 set2
SUNION key [key ...] 并集 SUNION set1 set2
SDIFF key [key ...] 差集 SDIFF set1 set2
SPOP key [count] 随机弹出元素 SPOP set1 2
SRANDMEMBER key [count] 随机获取但不移除 SRANDMEMBER set1 2

Python 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 给文章打标签
await redis_client.sadd("article:1:tags", "Python", "FastAPI", "后端")
await redis_client.sadd("article:2:tags", "Python", "数据分析", "机器学习")

# 判断文章是否有某标签
is_python = await redis_client.sismember("article:1:tags", "Python")   # True

# 获取所有标签
tags = await redis_client.smembers("article:1:tags")
# {"Python", "FastAPI", "后端"}

# 查找两篇文章的共同标签（交集）
common = await redis_client.sinter("article:1:tags", "article:2:tags")
# {"Python"}

# 两篇文章的标签并集
union = await redis_client.sunion("article:1:tags", "article:2:tags")

# 文章1独有标签（差集）
diff = await redis_client.sdiff("article:1:tags", "article:2:tags")
# {"FastAPI", "后端"}

5. Sorted Set（有序集合）

Sorted Set 在 Set 的基础上给每个元素绑定了分数（score），按分数排序。适合做排行榜、延时队列、限流器。

常用命令：

命令 说明 示例
ZADD key score member [score member ...] 添加元素（带分数） ZADD leaderboard 100 "豆"
ZRANGE key start stop [WITHSCORES] 按排名从低到高获取 ZRANGE leaderboard 0 9
ZREVRANGE key start stop [WITHSCORES] 按排名从高到低获取 ZREVRANGE leaderboard 0 9
ZSCORE key member 获取元素分数 ZSCORE leaderboard "豆"
ZRANK key member 获取元素排名（从低到高） ZRANK leaderboard "豆"
ZREVRANK key member 获取元素排名（从高到低） ZREVRANK leaderboard "豆"
ZINCRBY key increment member 增加元素的分数 ZINCRBY leaderboard 10 "豆"
ZREM key member [member ...] 移除元素 ZREM leaderboard "豆"
ZCARD key 获取元素数量 ZCARD leaderboard
ZCOUNT key min max 统计分数区间内元素数量 ZCOUNT leaderboard 80 100

ZRANGE 和 ZREVRANGE 的 start 和 stop 是排名索引（从 0 开始），不是分数。

Python 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 初始化排行榜
await redis_client.zadd("leaderboard", {
    "豆棍度子": 95,
    "曹操": 88,
    "曹植": 92,
    "曹雪芹": 78,
    "刘备": 85,
})

# 获取前三名（从高到低）
top3 = await redis_client.zrevrange("leaderboard", 0, 2, withscores=True)
# [("豆棍度子", 95.0), ("曹植", 92.0), ("曹操", 88.0)]

# 某人加分
await redis_client.zincrby("leaderboard", 10, "曹雪芹")   # 78 → 88

# 查看某人的排名（从高到低，0 表示第1名）
rank = await redis_client.zrevrank("leaderboard", "曹雪芹")  # 并列第3？

# 查看某人的分数
score = await redis_client.zscore("leaderboard", "曹植")    # 92.0

# 统计及格人数（60分以上）
count = await redis_client.zcount("leaderboard", 60, 100)

# 获取总分排名区间内的人（分数范围）
# 比如显示所有 80~100 分的人
high_scores = await redis_client.zrangebyscore("leaderboard", 80, 100, withscores=True)

过期时间与内存淘汰

设置过期时间

Redis 中几乎所有数据都可以设置过期时间，过期后自动删除。这是 Redis 作为缓存最重要的特性之一。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 方式一：SET 时直接指定
await redis_client.setex("cache:key", 3600, "value")        # 3600 秒后过期

# 方式二：设置后单独指定过期（秒）
await redis_client.set("cache:key", "value")
await redis_client.expire("cache:key", 3600)

# 方式三：指定过期时间点（datetime）
from datetime import datetime, timedelta
await redis_client.expireat("cache:key", datetime.now() + timedelta(hours=1))

# 查看剩余时间（秒）
ttl = await redis_client.ttl("cache:key")    # -1 表示未设置过期，-2 表示键已不存在

# 取消过期时间（永不过期）
await redis_client.persist("cache:key")

关键区别： TTL 返回 -1 表示键存在但没有过期时间，-2 表示键已不存在。

内存淘汰策略

当 Redis 内存写满时，根据配置的 maxmemory-policy 决定淘汰哪些数据：

策略 说明
noeviction 不淘汰，写入直接报错（默认）
allkeys-lru 淘汰最近最少使用的 key（最常用）
allkeys-lfu 淘汰使用频率最低的 key（Redis 4.0+）
volatile-lru 只在设置了过期时间的 key 中淘汰 LRU
volatile-ttl 淘汰即将过期的 key
allkeys-random 随机淘汰

生产环境推荐 allkeys-lru 或 allkeys-lfu：

1
2
3
# redis.conf 配置
maxmemory 1gb
maxmemory-policy allkeys-lru

在 FastAPI 中集成 Redis

缓存装饰器（完整示例）

把常用的缓存逻辑封装成装饰器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from functools import wraps
import json
from typing import Any, Callable
from fastapi import FastAPI
from redis.asyncio import Redis

redis_client: Redis | None = None

def cache(ttl: int = 300):
    """Redis 缓存装饰器"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 用函数名 + 参数生成缓存 key
            cache_key = f"cache:{func.__name__}:{json.dumps(kwargs, ensure_ascii=False)}"

            # 尝试从缓存获取
            cached = await redis_client.get(cache_key)
            if cached:
                return json.loads(cached)

            # 执行原函数
            result = await func(*args, **kwargs)

            # 写入缓存
            result_str = json.dumps(result, ensure_ascii=False, default=str)
            await redis_client.setex(cache_key, ttl, result_str)

            return result
        return wrapper
    return decorator

# 使用缓存装饰器
@app.get("/users/{user_id}")
@cache(ttl=60)   # 缓存 60 秒
async def get_user(user_id: int):
    # 模拟查库耗时
    import asyncio
    await asyncio.sleep(2)
    return {"user_id": user_id, "username": "豆棍度子", "age": 18}

缓存穿透保护（防止恶意攻击）

缓存穿透：请求的数据既不在缓存中，也不在数据库中，每次请求都直接打到数据库：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    cache_key = f"user:{user_id}"

    # 1. 查缓存
    cached = await redis_client.get(cache_key)
    if cached == "null":
        return {"error": "用户不存在"}          # 空缓存命中
    if cached:
        return json.loads(cached)

    # 2. 查数据库
    user = await fake_query_db(user_id)
    if not user:
        # 缓存空值，防止穿透（短时间即可，比如 60 秒）
        await redis_client.setex(cache_key, 60, "null")
        return {"error": "用户不存在"}

    # 3. 写入缓存
    await redis_client.setex(cache_key, 300, json.dumps(user, default=str))
    return user

核心思路：数据库返回空时，也在缓存中存一个空标记（TTL 设短一些），下次同样的请求直接拦在缓存层。

缓存雪崩预防

缓存雪崩：大量缓存同时过期，或 Redis 宕机，导致请求全部打回数据库。

1
2
3
4
5
6
7
# 方案一：过期时间加随机偏移，避免同时过期
import random

@app.get("/users/{user_id}")
@cache(ttl=300 + random.randint(0, 60))    # 基本过期时间 + 0~60 秒偏移
async def get_user(user_id: int):
    ...

接口限流（防刷）

使用 Redis 计数器 + 过期时间实现简单的 IP 限流：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import HTTPException, Request

@app.get("/api/data")
async def api_data(request: Request):
    ip = request.client.host
    key = f"rate_limit:{ip}"
    limit = 10           # 最多 10 次
    window = 60          # 60 秒内

    count = await redis_client.get(key)
    if count and int(count) >= limit:
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")

    await redis_client.incr(key, 1)
    await redis_client.expire(key, window)   # 刷新过期时间
    return {"data": "这里是接口返回的数据"}

更严谨的做法是用滑动窗口算法（使用 Sorted Set + ZREMRANGEBYSCORE）。

连接池与性能优化

为什么需要连接池

每次操作都创建/关闭连接开销很大（TCP 握手），连接池复用已有连接，大幅提高性能。

1
2
3
4
5
6
7
8
9
# 默认已有连接池
redis_client = Redis.from_url(
    "redis://localhost:6379/0",
    max_connections=20,      # 连接池大小
    decode_responses=True,
)

# 发送 PING 检测连接是否正常
await redis_client.ping()   # True

Pipeline：批量操作

Pipeline 可以将多条命令打包一次发送，减少网络往返（RTT）：

1
2
3
4
5
6
7
8
9
# ❌ 逐条发送（性能差）
for i in range(1000):
    await redis_client.set(f"key:{i}", i)

# ✅ 使用 Pipeline 批量发送（快 ≈100 倍）
async with redis_client.pipeline() as pipe:
    for i in range(1000):
        pipe.set(f"key:{i}", i)
    await pipe.execute()

Pipeline 不会立即执行，直到调用 execute() 才一起发送。适合不依赖中间结果的批量操作。

项目实战：Redis 工具类封装

把常用的 Redis 操作封装成一个服务类，路由里直接调：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import json
from typing import Any
from redis.asyncio import Redis

class RedisService:
    """Redis 业务封装"""

    def __init__(self, redis: Redis):
        self.redis = redis

    # ---- 缓存操作 ----

    async def get_cache(self, key: str) -> Any | None:
        """获取缓存（自动解析 JSON）"""
        data = await self.redis.get(key)
        return json.loads(data) if data else None

    async def set_cache(self, key: str, value: Any, ttl: int = 300):
        """设置缓存（自动序列化 JSON）"""
        data = json.dumps(value, ensure_ascii=False, default=str)
        await self.redis.setex(key, ttl, data)

    # ---- 计数器 ----

    async def increment(self, key: str, amount: int = 1) -> int:
        """自增计数"""
        return await self.redis.incrby(key, amount)

    async def get_count(self, key: str) -> int:
        """获取计数"""
        result = await self.redis.get(key)
        return int(result) if result else 0

    # ---- 限流 ----

    async def check_rate_limit(self, key: str, max_count: int, window: int) -> bool:
        """检查是否超过限流阈值，返回 True 表示允许通过"""
        count = await self.redis.get(key)
        if count and int(count) >= max_count:
            return False

        await self.redis.incr(key, 1)
        await self.redis.expire(key, window)
        return True

    # ---- 缓存穿透防护 ----

    async def safe_get(self, key: str, db_func, ttl: int = 300, empty_ttl: int = 60):
        """缓存穿透防护版查询"""
        cached = await self.redis.get(key)
        if cached == "__NULL__":
            return None
        if cached:
            return json.loads(cached)

        # 查数据库
        result = await db_func()
        if result is None:
            await self.redis.setex(key, empty_ttl, "__NULL__")     # 缓存空值
            return None

        await self.redis.setex(key, ttl, json.dumps(result, default=str))
        return result

    # ---- 分布式锁（简单版） ----

    async def acquire_lock(self, lock_key: str, ttl: int = 10) -> bool:
        """获取分布式锁"""
        return bool(await self.redis.setnx(lock_key, "locked")) and await self.redis.expire(lock_key, ttl)

    async def release_lock(self, lock_key: str):
        """释放分布式锁"""
        await self.redis.delete(lock_key)


# 在 FastAPI 中注入
async def get_redis_service() -> RedisService:
    return RedisService(redis_client)

@app.get("/users/{user_id}")
async def get_user(user_id: int, redis_svc: RedisService = Depends(get_redis_service)):
    user = await redis_svc.safe_get(
        key=f"user:{user_id}",
        db_func=lambda: fake_query_db(user_id),
        ttl=300,
    )
    if not user:
        return {"error": "用户不存在"}
    return user

速查表

数据结构速查

结构 说明 典型应用 底层实现
String 字符串 / 数字 缓存、计数器、验证码 动态字符串
Hash 键值对集合（对象） 用户信息、商品信息 哈希表
List 双向链表 消息队列、最新列表 快速列表
Set 无序唯一集合 标签、去重、共同好友 哈希表
Sorted Set 有序唯一集合（带分数） 排行榜、延时队列 跳表 + 哈希表

Redis 命令速查

操作 String Hash List Set Sorted Set
增 / 改 SET HSET LPUSH / RPUSH SADD ZADD
查 GET HGET / HGETALL LRANGE SMEMBERS ZRANGE / ZREVRANGE
删 DEL HDEL LPOP / RPOP SREM ZREM
是否存在 EXISTS HEXISTS - SISMEMBER -
长度 / 计数 STRLEN HLEN LLEN SCARD ZCARD
原子增减 INCR / DECR HINCRBY - - ZINCRBY
过期时间 EXPIRE / TTL 同上 同上 同上 同上

最佳实践总结

要点 说明
异步连接 FastAPI 中必须用 redis.asyncio.Redis
连接池 设置 max_connections，避免频繁创建/销毁连接
解码设置 decode_responses=True，避免手动处理 bytes
key 命名规范 用冒号分层：user:1001:profile，cache:article:42
设置过期时间 所有缓存数据必须设置 TTL，防止内存爆满
缓存穿透防护 空结果也缓存（短 TTL）
缓存雪崩防护 过期时间加随机偏移
批量操作用 Pipeline 减少网络往返
敏感数据不要存 Redis Redis 默认无认证，不建议存放密码、Token 等
生产环境设置密码 requirepass 配置 + AUTH 命令
避免大 Key 单个 key 的 value 不要超过 10MB，会影响集群迁移和读写性能