Redis 完全指南：从基础数据结构到集群部署与性能优化 | 2026最新实践

1
# Python 示例
2
import redis
3

4
r = redis.Redis(host='localhost', port=6379, db=0)
5

6
# 基本操作
7
r.set('name', 'Alice', ex=3600)  # 带过期时间
8
name = r.get('name')  # b'Alice'
9

10
# 计数器
11
r.set('counter', 100)
12
r.incr('counter')        # 101
13
r.incrby('counter', 10)  # 111

1
# Python 示例
2
r.hset('user:1', mapping={
3
    'name': 'Alice',
4
    'age': 30,
5
    'email': 'alice@example.com'
6
})
7
user = r.hgetall('user:1')
8
# {b'name': b'Alice', b'age': b'30', b'email': b'alice@example.com'}

1
# Python 示例：游戏排行榜
2
r.zadd('game_ranking', {
3
    'Alice': 100,
4
    'Bob': 85,
5
    'Charlie': 92
6
})
7

8
# 获取 Top 3
9
top3 = r.zrevrange('game_ranking', 0, 2, withscores=True)
10
# [(b'Alice', 100.0), (b'Charlie', 92.0), (b'Bob', 85.0)]
11

12
# 增加分数
13
r.zincrby('game_ranking', 10, 'Bob')  # Bob: 95

1
# Python 发布者
2
r.publish('news', 'Hello World')
3

4
# Python 订阅者
5
pubsub = r.pubsub()
6
pubsub.subscribe('news')
7
for message in pubsub.listen():
8
    if message['type'] == 'message':
9
        print(message['data'])  # b'Hello World'

1
# Python 执行 Lua 脚本
2
lua_script = """
3
if redis.call('get', KEYS[1]) == ARGV[1] then
4
    return redis.call('del', KEYS[1])
5
else
6
    return 0
7
end
8
"""
9

10
# 分布式锁释放
11
r.eval(lua_script, 1, 'lock_key', 'unique_value')

1
# Python Stream 示例
2
# 生产
3
r.xadd('events', {'name': 'Alice', 'action': 'login'})
4

5
# 消费
6
r.xread({'events': '0'}, count=10)

1
主从复制架构：
2

3
    Master (读写)
4
       │
5
    ┌──┴──┐
6
    │     │
7
  Slave1  Slave2 (只读)
8

9
特点：
10
- 读写分离：主节点写，从节点读
11
- 数据冗余：从节点保存数据副本
12
- 故障恢复：主节点故障可手动切换

1
哨兵模式架构：
2

3
  Sentinel1    Sentinel2    Sentinel3
4
       │            │            │
5
       └────────────┼────────────┘
6
                    │
7
            ┌───────┴───────┐
8
            │               │
9
         Master           Slave
10

11
哨兵职责：
12
1. 监控：检测主从节点是否存活
13
2. 通知：通知运维人员故障
14
3. 自动故障转移：主节点故障时自动选举新主
15
4. 配置提供：客户端连接哨兵获取主节点地址

1
Redis Cluster 架构（6个节点，3主3从）：
2

3
节点A (master)  ←→  节点B (slave)
4
  slot 0-5460
5

6
节点C (master)  ←→  节点D (slave)
7
  slot 5461-10922
8

9
节点E (master)  ←→  节点F (slave)
10
  slot 10923-16383
11

12
特点：
13
- 数据分片：16384 个 slot 分配到各节点
14
- 去中心化：节点间通过 Gossip 协议通信
15
- 自动故障转移：主节点故障，从节点升级
16
- 水平扩展：可动态添加/移除节点

1
# 问题：查询不存在的数据，绕过缓存直接打到数据库
2

3
# 方案1：缓存空值
4
def get_user(user_id):
5
    cache_key = f"user:{user_id}"
6
    user = r.get(cache_key)
7

8
    if user is not None:
9
        if user == b'':
10
            return None  # 缓存的空值
11
        return json.loads(user)
12

13
    # 查询数据库
14
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
15

16
    if user is None:
17
        # 缓存空值，短过期时间
18
        r.setex(cache_key, 300, '')
19
    else:
20
        r.setex(cache_key, 3600, json.dumps(user))
21

22
    return user
23

24
# 方案2：布隆过滤器
25
from redisbloom.client import Client
26

27
rb = Client()
28
rb.bfCreate('users_filter', 0.01, 1000000)  # 错误率1%，容量100万
29

30
# 添加存在的用户ID
31
rb.bfAdd('users_filter', 'user_123')
32

33
def get_user_safe(user_id):
34
    # 先检查布隆过滤器
35
    if not rb.bfExists('users_filter', user_id):
36
        return None  # 一定不存在
37

38
    # 再查缓存和数据库
39
    return get_user(user_id)

1
# 问题：热点key过期瞬间，大量请求同时打到数据库
2

3
# 方案：互斥锁
4
import time
5
import uuid
6

7
def get_hot_data(key):
8
    # 先查缓存
9
    data = r.get(key)
10
    if data is not None:
11
        return json.loads(data)
12

13
    # 获取互斥锁
14
    lock_key = f"lock:{key}"
15
    lock_value = str(uuid.uuid4())
16

17
    # 尝试获取锁（设置过期时间防止死锁）
18
    if r.set(lock_key, lock_value, nx=True, ex=10):
19
        try:
20
            # 查询数据库
21
            data = db.query_hot_data(key)
22
            # 写入缓存
23
            r.setex(key, 3600, json.dumps(data))
24
            return data
25
        finally:
26
            # 释放锁（Lua脚本保证原子性）
27
            lua = """
28
            if redis.call('get', KEYS[1]) == ARGV[1] then
29
                return redis.call('del', KEYS[1])
30
            else
31
                return 0
32
            end
33
            """
34
            r.eval(lua, 1, lock_key, lock_value)
35
    else:
36
        # 等待后重试
37
        time.sleep(0.1)
38
        return get_hot_data(key)

1
# 问题：大量key同时过期，请求全部打到数据库
2

3
# 方案1：过期时间加随机值
4
def cache_data(key, data, base_ttl=3600):
5
    import random
6
    # 过期时间加 0-300 秒随机值
7
    ttl = base_ttl + random.randint(0, 300)
8
    r.setex(key, ttl, json.dumps(data))
9

10
# 方案2：多级缓存
11
import functools
12

13
@functools.lru_cache(maxsize=1000)  # 本地缓存
14
def get_data_l1(key):
15
    # L2: Redis 缓存
16
    data = r.get(key)
17
    if data:
18
        return json.loads(data)
19

20
    # L3: 数据库
21
    data = db.query(key)
22
    if data:
23
        cache_data(key, data)
24
    return data

1
# 不推荐：逐条执行
2
for i in range(1000):
3
    r.set(f'key:{i}', i)
4
# 1000 次网络往返
5

6
# 推荐：Pipeline
7
pipe = r.pipeline()
8
for i in range(1000):
9
    pipe.set(f'key:{i}', i)
10
pipe.execute()
11
# 1 次网络往返
12

13
# 事务型 Pipeline
14
pipe = r.pipeline(transaction=True)
15
pipe.set('key1', 'value1')
16
pipe.incr('counter')
17
pipe.execute()

1
import redis
2
import json
3

4
# 连接池
5
pool = redis.ConnectionPool(
6
    host='localhost',
7
    port=6379,
8
    db=0,
9
    max_connections=20,
10
    decode_responses=True  # 自动解码为字符串
11
)
12

13
r = redis.Redis(connection_pool=pool)
14

15
# 基本操作
16
r.set('key', 'value', ex=3600)
17
value = r.get('key')
18

19
# JSON 缓存
20
def cache_json(key, data, ttl=3600):
21
    r.setex(key, ttl, json.dumps(data, ensure_ascii=False))
22

23
def get_json(key):
24
    data = r.get(key)
25
    return json.loads(data) if data else None
26

27
# 分布式锁
28
def acquire_lock(lock_name, timeout=10):
29
    identifier = str(uuid.uuid4())
30
    end = time.time() + timeout
31
    while time.time() < end:
32
        if r.set(lock_name, identifier, nx=True, ex=timeout):
33
            return identifier
34
        time.sleep(0.001)
35
    return None
36

37
def release_lock(lock_name, identifier):
38
    lua = """
39
    if redis.call('get', KEYS[1]) == ARGV[1] then
40
        return redis.call('del', KEYS[1])
41
    else
42
        return 0
43
    end
44
    """
45
    return r.eval(lua, 1, lock_name, identifier)

1
const Redis = require('ioredis');
2

3
// 连接集群
4
const redis = new Redis.Cluster([
5
    { host: '127.0.0.1', port: 7000 },
6
    { host: '127.0.0.1', port: 7001 },
7
    { host: '127.0.0.1', port: 7002 },
8
]);
9

10
// 基本操作
11
await redis.set('key', 'value', 'EX', 3600);
12
const value = await redis.get('key');
13

14
// Pipeline
15
const pipeline = redis.pipeline();
16
for (let i = 0; i < 1000; i++) {
17
    pipeline.set(`key:${i}`, i);
18
}
19
const results = await pipeline.exec();
20

21
// 事务
22
const multi = redis.multi();
23
multi.set('key1', 'value1');
24
multi.incr('counter');
25
const results = await multi.exec();

1
小型项目：单机 Redis（开启持久化）
2
中型项目：主从 + 哨兵（高可用）
3
大型项目：Cluster 集群（水平扩展）
4
缓存场景：可关闭持久化，使用最大内存策略
5
数据重要：RDB + AOF 混合持久化