Redis 完全指南 2026：核心数据结构 + 缓存设计模式 + 持久化 + 高可用集群

1
import redis.asyncio as aioredis
2
import json
3

4
redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)
5

6
async def get_user(user_id: int) -> dict | None:
7
    cache_key = f"user:{user_id}"
8

9
    # 1. 查缓存
10
    cached = await redis_client.get(cache_key)
11
    if cached:
12
        return json.loads(cached)
13

14
    # 2. 缓存未命中，查数据库
15
    user = await db.fetch_one("SELECT * FROM users WHERE id = ?", user_id)
16
    if user is None:
17
        # 缓存空值，防止穿透（短 TTL）
18
        await redis_client.set(cache_key, "null", ex=60)
19
        return None
20

21
    # 3. 写入缓存
22
    await redis_client.set(cache_key, json.dumps(dict(user)), ex=3600)
23
    return dict(user)
24

25
async def update_user(user_id: int, data: dict) -> None:
26
    await db.execute("UPDATE users SET ... WHERE id = ?", user_id)
27
    # 更新数据库后删缓存（下次读时重建），不要直接更新缓存！
28
    await redis_client.delete(f"user:{user_id}")

1
import random
2

3
async def set_cache(key: str, value: str, base_ttl: int = 3600):
4
    # 加随机偏移（±10%），避免大量 key 同时过期
5
    jitter = random.randint(-base_ttl // 10, base_ttl // 10)
6
    await redis_client.set(key, value, ex=base_ttl + jitter)

1
import asyncio, uuid
2

3
async def get_hot_data(key: str) -> dict:
4
    cached = await redis_client.get(key)
5
    if cached:
6
        return json.loads(cached)
7

8
    lock_key = f"lock:{key}"
9
    lock_val = str(uuid.uuid4())
10

11
    # NX：不存在才设置（只有一个请求拿到锁）
12
    acquired = await redis_client.set(lock_key, lock_val, nx=True, ex=10)
13
    if acquired:
14
        try:
15
            cached = await redis_client.get(key)   # 双重检查
16
            if cached:
17
                return json.loads(cached)
18

19
            data = await db.fetch_data(key)
20
            await redis_client.set(key, json.dumps(data), ex=3600)
21
            return data
22
        finally:
23
            # Lua 脚本原子释放锁（只释放自己的锁）
24
            script = """if redis.call("get",KEYS[1])==ARGV[1] then
25
                return redis.call("del",KEYS[1]) else return 0 end"""
26
            await redis_client.eval(script, 1, lock_key, lock_val)
27
    else:
28
        await asyncio.sleep(0.1)
29
        return await get_hot_data(key)

1
# 初始化布隆过滤器（100万元素，误判率 0.1%）
2
await redis_client.execute_command("BF.RESERVE", "user_bloom", 0.001, 1_000_000)
3

4
async def create_user(user_id: int, data: dict):
5
    await db.insert(user_id, data)
6
    # 同步到布隆过滤器
7
    await redis_client.execute_command("BF.ADD", "user_bloom", user_id)
8

9
async def get_user_safe(user_id: int) -> dict | None:
10
    # 布隆过滤器说不存在 → 一定不存在，直接返回
11
    if not await redis_client.execute_command("BF.EXISTS", "user_bloom", user_id):
12
        return None
13
    return await get_user(user_id)

1
主从架构：Master → Slave × 2
2
哨兵监控：Sentinel × 3（奇数，防脑裂）
3
主节点故障 → 哨兵选举 → 从节点升主 → 客户端更新连接

1
from redis.sentinel import Sentinel
2

3
sentinel = Sentinel(
4
    [("sentinel1", 26379), ("sentinel2", 26379), ("sentinel3", 26379)],
5
    socket_timeout=0.1,
6
)
7
master = sentinel.master_for("mymaster")   # 写操作
8
slave  = sentinel.slave_for("mymaster")    # 读操作（负载均衡）

1
最小配置：3主节点（16384个哈希槽均分）+ 每主1从
2
数据路由：客户端计算 HASH_SLOT(key) = CRC16(key) % 16384

1
from redis.cluster import RedisCluster
2

3
client = RedisCluster(
4
    host="127.0.0.1", port=7000,
5
    decode_responses=True,
6
    skip_full_coverage_check=True,
7
)

1
async def is_rate_limited(user_id: str, limit: int = 100, window: int = 60) -> bool:
2
    key = f"ratelimit:{user_id}"
3
    now = time.time()
4

5
    pipe = redis_client.pipeline()
6
    pipe.zremrangebyscore(key, "-inf", now - window)       # 删除窗口外记录
7
    pipe.zadd(key, {str(uuid.uuid4()): now})               # 记录本次请求
8
    pipe.zcard(key)                                        # 窗口内请求数
9
    pipe.expire(key, window + 1)
10
    results = await pipe.execute()
11

12
    return results[2] > limit     # 超出限制返回 True

1
async def schedule_task(task_data: dict, delay: int) -> None:
2
    """delay 秒后执行"""
3
    await redis_client.zadd(
4
        "delayed_queue",
5
        {json.dumps(task_data): time.time() + delay}
6
    )
7

8
async def poll_delayed_tasks():
9
    while True:
10
        now = time.time()
11
        tasks = await redis_client.zrangebyscore("delayed_queue", "-inf", now)
12
        for task_json in tasks:
13
            if await redis_client.zrem("delayed_queue", task_json):  # 原子删除防重复
14
                await process_task(json.loads(task_json))
15
        await asyncio.sleep(0.5)

1
async def create_session(user_id: int) -> str:
2
    sid = str(uuid.uuid4())
3
    await redis_client.setex(f"session:{sid}", 86400, json.dumps({"user_id": user_id}))
4
    return sid
5

6
async def get_session(sid: str) -> dict | None:
7
    data = await redis_client.get(f"session:{sid}")
8
    if not data:
9
        return None
10
    await redis_client.expire(f"session:{sid}", 86400)  # 每次访问刷新 TTL
11
    return json.loads(data)