Redis 是目前最流行的内存数据库，广泛应用于缓存、消息队列、排行榜等场景。 本文将深入 Redis 的核心概念，从基础数据结构到生产环境的高可用集群，提供完整的实践指南。

核心数据结构与应用场景

String — 最基础的类型

# 缓存对象（JSON 序列化）
SET user:1001 '{"name":"Alice","age":28}' EX 3600

# 计数器
INCR article:view:20260710
INCRBY user:1001:score 10

# 分布式锁
SET lock:order:1001 unique_value NX EX 30

Hash — 存储对象

# 相比 String+JSON，Hash 可以按字段读取，节省带宽
HSET user:1001 name "Alice" age 28 email "alice@example.com"
HGET user:1001 name       # 只读取 name
HINCRBY user:1001 age 1   # 原子递增

List — 消息队列

# 生产者-消费者队列
LPUSH task_queue "send_email:1001"
LPUSH task_queue "generate_report:2001"
BRPOP task_queue 5  # 阻塞等待，超时 5 秒

# 最新动态列表（保留最近 100 条）
LPUSH user:timeline "动态内容"
LTRIM user:timeline 0 99

Set — 去重与交并集

# 共同关注
SADD user:1:follows 101 102 103 104
SADD user:2:follows 102 103 105 106
SINTER user:1:follows user:2:follows  # 共同关注: {102, 103}

Sorted Set — 排行榜

# 文章排行榜（按阅读量）
ZADD article:ranking 2340 "文章A" 1890 "文章B" 3120 "文章C"
ZREVRANGE article:ranking 0 9 WITHSCORES  # Top 10
ZINCRBY article:ranking 1 "文章A"         # 阅读量+1

缓存策略

Cache-Aside（旁路缓存）

最常用的缓存模式：读时先查缓存，未命中则查数据库并回写缓存；写时更新数据库后删除缓存。

def get_user(user_id: int):
    # 1. 先查缓存
    key = f"user:{user_id}"
    cached = redis.get(key)
    if cached:
        return json.loads(cached)

    # 2. 缓存未命中，查数据库
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user:
        # 3. 回写缓存，加随机过期时间防雪崩
        redis.setex(key, 3600 + random.randint(0, 300), json.dumps(user))
    return user

def update_user(user_id: int, data: dict):
    # 1. 更新数据库
    db.execute("UPDATE users SET ... WHERE id = %s", user_id)
    # 2. 删除缓存（而非更新）
    redis.delete(f"user:{user_id}")

缓存三大问题

• 缓存穿透：查询不存在的数据，请求直达数据库。解决方案：布隆过滤器或缓存空值。
• 缓存击穿：热点 Key 过期瞬间大量请求涌入数据库。解决方案：互斥锁或"永不过期"。
• 缓存雪崩：大量 Key 同时过期。解决方案：过期时间加随机值、多级缓存。

# 缓存穿透：缓存空值
def get_user_safe(user_id: int):
    key = f"user:{user_id}"
    cached = redis.get(key)
    if cached is not None:
        return json.loads(cached) if cached != "NULL" else None
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user:
        redis.setex(key, 3600, json.dumps(user))
    else:
        redis.setex(key, 60, "NULL")  # 缓存空值 60 秒
    return user

# 缓存击穿：互斥锁
def get_user_mutex(user_id: int):
    key = f"user:{user_id}"
    cached = redis.get(key)
    if cached:
        return json.loads(cached)

    lock_key = f"lock:user:{user_id}"
    if redis.set(lock_key, "1", nx=True, ex=10):
        try:
            user = db.query("SELECT * FROM users WHERE id = %s", user_id)
            if user:
                redis.setex(key, 3600, json.dumps(user))
            return user
        finally:
            redis.delete(lock_key)
    else:
        time.sleep(0.1)
        return get_user_mutex(user_id)

持久化：RDB vs AOF

高可用架构

主从复制：一主多从，读写分离，数据异步复制。

哨兵模式：在主从基础上增加哨兵进程，自动监控、故障转移。

Redis Cluster：去中心化的分布式方案，数据自动分片，支持水平扩展。

# Sentinel 配置示例 (sentinel.conf)
sentinel monitor mymaster 192.168.1.10 6379 2
sentinel down-after-milliseconds mymaster 5000
sentinel parallel-syncs mymaster 1
sentinel failover-timeout mymaster 15000

总结

掌握 Redis 不仅是会用几个命令，更重要的是理解数据结构的选择、缓存策略的设计 和高可用架构的规划。在面试和实际项目中，缓存击穿/穿透/雪崩的解决方案、 RDB 与 AOF 的选择、Cluster 分片原理都是高频考察点。