前言

Redis 是目前最流行的内存数据库。几乎所有后端系统里都能看到它的身影——做缓存、做分布式锁、做排行榜、做消息队列。它把”快”做到了极致（单机 10 万 QPS 是常态），同时又提供了丰富的数据结构。

这篇文章覆盖 Redis 最核心的数据结构和实际开发中最常用的几个场景。

一、Redis 是什么

简单说，Redis 是一个基于内存的键值存储系统。和 MySQL 的区别在于：

MySQL Redis
存储位置 硬盘 内存
速度 毫秒级 微秒级
数据结构 表（行列） String、Hash、List、Set、Sorted Set
持久化 天然持久 可选（RDB 快照 / AOF 日志）
典型用途 持久存储业务数据 缓存、临时数据、实时计算

内存意味着快，但也意味着数据在重启后会丢失（除非开了持久化）。所以 Redis 通常不用于存”丢了就没了”的核心业务数据，而是作为 MySQL 前面的缓存层或临时数据存储。

二、安装

1
2
3
4
5
6
7
8
9
10
# macOS
brew install redis
brew services start redis

# Ubuntu
sudo apt install redis-server
sudo systemctl start redis

# Docker（推荐开发环境用）
docker run -d --name redis -p 6379:6379 redis:7-alpine

连接验证：

1
2
3
redis-cli
127.0.0.1:6379> PING
PONG

三、五大数据结构

3.1 String（字符串）

最基础的类型，也是缓存最常用的类型。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
SET user:1:name "July"
GET user:1:name                    # "July"

# 带过期时间的设置（缓存场景核心命令）
SETEX verification_code:13800138000 300 "123456"   # 300 秒后自动删除

# 数值操作（计数器）
SET article:42:views 100
INCR article:42:views              # 101
INCRBY article:42:views 10         # 111

# 分布式锁的基础
SET lock:order:20260530001 "holder" NX EX 30
# NX: key 不存在才设置成功，EX 30: 30 秒后自动释放

典型场景：

1
2
3
4
缓存 JSON → SET user:1 '{"name":"July","age":25}' EX 3600
计数器   → INCR page:home:visits
分布式锁 → SET lock:resource "owner" NX EX 30
限流     → INCR rate:api:user:1 配合 EXPIRE

3.2 Hash（哈希表）

存对象的首选，比把 JSON 序列化成 String 更省空间、支持部分字段读写。

1
2
3
4
5
6
7
HSET user:1 name "July" age 25 city "深圳"
HGET user:1 name                   # "July"
HGETALL user:1                     # 所有字段
HINCRBY user:1 age 1               # age 26

# 只更新一个字段，不影响其他字段（比存 JSON String 方便得多）
HSET user:1 email "new@example.com"

3.3 List（列表）

有序、可重复的双向链表，天然适合做队列和栈。

1
2
3
4
5
6
7
8
9
10
11
12
LPUSH queue:emails "email1" "email2" "email3"     # 左侧入队
RPOP queue:emails                                   # 右侧出队：email1

LLEN queue:emails                                   # 队列长度

# 阻塞读取（消费者等待新消息，超时 5 秒）
BRPOP queue:emails 5

# 最近 N 条记录的存储（如最近 10 条登录日志）
LPUSH login_logs "2026-05-30 14:00:01 user July登录"
LTRIM login_logs 0 9           # 只保留前 10 条
LRANGE login_logs 0 -1         # 查看全部

3.4 Set（集合）

无序、不重复，擅长交集并集差集运算。

1
2
3
4
5
6
7
8
9
10
11
SADD article:1:tags "Redis" "数据库" "缓存"
SADD article:2:tags "Redis" "分布式锁"
SISMEMBER article:1:tags "Redis"     # 1（存在）

# 两个集合的交集（找出同时有两类标签的文章）
SINTER article:1:tags article:2:tags  # "Redis"

# 共同关注、共同好友等社交场景
SADD user:1:followers "user2" "user3" "user4"
SADD user:2:followers "user3" "user4" "user5"
SINTER user:1:followers user:2:followers  # 共同关注

3.5 Sorted Set（有序集合）

每个成员带一个分数，自动按分数排序。最常用的场景是排行榜。

1
2
3
4
5
6
7
8
9
10
11
12
13
ZADD leaderboard 100 "player1" 200 "player2" 150 "player3"

# 分数增加（玩家得分更新）
ZINCRBY leaderboard 50 "player1"

# 按分数降序排列（排行榜 Top 10）
ZREVRANGE leaderboard 0 9 WITHSCORES

# 查询某个玩家的排名（从 0 开始）
ZREVRANK leaderboard "player1"

# 查询分数在 150-300 之间的玩家
ZRANGEBYSCORE leaderboard 150 300 WITHSCORES

四、六大典型应用场景

场景 1：缓存（最常见）

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
import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_user(user_id):
    # 1. 先查缓存
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # 2. 缓存没有，查数据库
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    
    # 3. 写入缓存，过期时间 1 小时
    if user:
        r.setex(cache_key, 3600, json.dumps(user))
    
    return user

def update_user(user_id, data):
    # 先更新数据库
    db.execute(f"UPDATE users SET ... WHERE id = {user_id}")
    # 删除缓存（下次读取时自动重建）
    r.delete(f"user:{user_id}")

缓存穿透、击穿、雪崩：

问题 场景 解决方案
缓存穿透 查一个不存在的数据，每次都穿透到 DB 布隆过滤器，或把空结果也缓存（短过期时间）
缓存击穿 热点 key 过期瞬间大量请求打到 DB 互斥锁，只让一个请求去查 DB 并重建缓存
缓存雪崩 大量 key 同时过期 过期时间加随机偏移 EX 3600 + random(0,600)

场景 2：分布式锁

多台服务器竞争同一个资源时，需要一把”大家都能看到的锁”。Redis 的 SET NX EX 天然适合。

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
import time

def acquire_lock(lock_name, holder_id, expire_seconds=30):
    """获取分布式锁"""
    return r.set(lock_name, holder_id, nx=True, ex=expire_seconds)

def release_lock(lock_name, holder_id):
    """释放锁（用 Lua 脚本保证原子性）"""
    script = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    else
        return 0
    end
    """
    return r.eval(script, 1, lock_name, holder_id)

# 使用
lock_key = "lock:order:20260530001"
holder = f"server-3-{time.time()}"

if acquire_lock(lock_key, holder, 30):
    try:
        # 处理订单...
        process_order()
    finally:
        release_lock(lock_key, holder)

释放锁必须用 Lua 脚本的原因是：不能先 GET 再 DEL，这两步之间可能被别的线程插入——必须保证”判断 + 删除”是原子操作。

场景 3：排行榜

Sorted Set 是排行榜的完美数据结构，单机 Redis 能轻松支撑百万用户的实时排名。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 更新分数
def update_score(user_id, score):
    r.zadd("game:leaderboard", {user_id: score})

# Top 100
def get_top100():
    return r.zrevrange("game:leaderboard", 0, 99, withscores=True)

# 某个用户的排名和周榜前 10 名
def get_user_rank(user_id):
    rank = r.zrevrank("game:leaderboard", user_id)
    nearby = r.zrevrange("game:leaderboard", max(0, rank-5), rank+5, withscores=True)
    return rank, nearby

# 每周重置排行榜
r.delete("game:leaderboard")

场景 4：消息队列

List 的阻塞读可以当简易消息队列用。简单场景（无需消息确认、无需复杂路由）完全够用。

1
2
3
4
5
6
7
8
9
10
11
12
# 生产者
def enqueue_task(task_data):
    r.lpush("task:queue", json.dumps(task_data))

# 消费者
def worker():
    while True:
        result = r.brpop("task:queue", timeout=5)
        if result:
            _, task_json = result
            task = json.loads(task_json)
            process_task(task)

复杂场景建议用 RabbitMQ 或 Kafka，但 Redis 队列胜在零配置。

场景 5：Session 共享

多台 Web 服务器共享用户登录状态。

1
2
3
4
5
6
7
8
from flask import Flask, session
import redis
from datetime import timedelta

app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = redis.Redis(host='redis-server', port=6379)
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=2)

这样无论用户请求被分发到哪台服务器，都能读到同一个 Session。

场景 6：限流

防止某个用户或 IP 短时间内大量请求。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def rate_limit(user_id, max_requests=10, window_seconds=60):
    key = f"rate:{user_id}"
    current = r.get(key)
    
    if current is None:
        r.setex(key, window_seconds, 1)
        return True
    
    current = int(current)
    if current >= max_requests:
        return False
    
    r.incr(key)
    return True

# 每个用户每分钟最多 10 次请求
if not rate_limit(user_id, 10, 60):
    return "请求过于频繁，请稍后再试", 429

五、持久化策略

Redis 提供两种持久化方式，可以只开一种，也可以同时开。

方式 原理 优点 缺点
RDB 定时快照，把整个内存写入磁盘 恢复快、文件小 可能丢失最后一次快照后的数据
AOF 记录每一条写命令 数据更安全 文件大、恢复慢

配置（redis.conf）：

1
2
3
4
5
6
7
8
# RDB：900 秒内至少 1 次变更就生成快照
save 900 1
save 300 10
save 60 10000

# AOF：每条命令都记录
appendonly yes
appendfsync everysec   # 每秒刷盘一次

实际选择：对数据要求不高的缓存场景，不开持久化也没问题。需要持久化的场景，推荐同时开 RDB + AOF，用 RDB 做备份（恢复快），用 AOF 保证数据安全。

六、常见问题

Q：Redis 为什么这么快？

  1. 纯内存操作，没有磁盘 IO
  2. 单线程模型，没有锁切换开销
  3. IO 多路复用，一个线程处理大量连接
  4. 数据结构设计简单高效

Q：Redis 单线程为什么还这么快？
因为它的瓶颈不在 CPU，而在网络和内存带宽。单线程避免了上下文切换和锁竞争，反而效率更高。Redis 6.0 之后引入了多线程处理网络 IO，但命令执行还是单线程的。

Q：数据量超过内存怎么办？
Redis 不是用来存海量数据的。如果数据量大，用 Redis Cluster（分片）或者冷热分离（热数据在 Redis，全量在 MySQL）。

以上覆盖了 Redis 日常开发最常用的内容。实际项目中先从缓存用起，分布式锁、排行榜这些遇到具体场景再深入研究。