Redis大厂面试深度笔记

┌─────────────────────────────────────────┐
│              Redis 主线程               │
│                                         │
│  ┌───────────┐   ┌───────────────────┐  │
│  │ IO 多路    │   │   事件处理器        │  │
│  │ 复用程序    │──→│  (Event Handler)   │  │
│  │ (epoll)    │   │                   │  │
│  └───────────┘   │ ┌───────────────┐ │  │
│                   │ │ 连接应答处理器  │ │  │
│  ┌───────────┐    │ │ (acceptTcp)   │ │  │
│  │  事件队列   │    │ ├───────────────┤ │  │
│  │ (fd+mask)  │    │ │ 命令请求处理器  │ │  │
│  └───────────┘    │ │ (readQuery)   │ │  │
│                   │ ├───────────────┤ │  │
│                   │ │ 命令回复处理器  │ │  │
│                   │ │ (sendReply)   │ │  │
│                   │ ├───────────────┤ │  │
│                   │ │ 定时事件处理器  │ │  │
│                   │ │ (TimeEvent)   │ │  │
│                   │ └───────────────┘ │  │
│                   └───────────────────┘  │
└─────────────────────────────────────────┘

// src/server.c —— Redis 服务器启动入口
int main() {
    initServer();      // 创建事件循环、注册事件处理器
    while(server.shutdown == 0) {
        // aeProcessEvents 是事件循环核心
        aeProcessEvents();
    }
}

// src/ae.c —— 事件循环核心
int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
    // 1. 计算最近定时任务的等待时间
    shortest = aeSearchNearestTimer();
    // 2. 调用 epoll_wait 等待事件，阻塞最短 shortest 时间
    numevents = aeApiPoll(eventLoop, tvp);
    // 3. 处理所有就绪的 IO 事件
    for (j = 0; j < numevents; j++) {
        aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
        if (fe->mask & AE_READABLE) fe->rfileProc(...);  // 读事件 → 命令请求处理器
        if (fe->mask & AE_WRITABLE) fe->wfileProc(...);  // 写事件 → 命令回复处理器
    }
    // 4. 处理定时事件（serverCron 等）
    processed += processTimeEvents();
}

客户端发送命令
  ↓
epoll 返回可读事件
  ↓
readQueryFromClient() —— 读取客户端命令到输入缓冲区
  ↓
processInputBuffer() —— 解析协议（RESP）
  ↓
processCommand() —— 查命令表，校验，执行
  ↓
addReply() —— 将结果写入输出缓冲区
  ↓
epoll 返回可写事件
  ↓
writeToClient() —— 将缓冲区数据发送给客户端

┌──────────┐     ┌──────────────────────────────┐
│  客户端   │     │        Redis 进程             │
│ 连接1    │────→│                              │
│ 连接2    │────→│  ┌─────────┐  ┌───────────┐  │
│ 连接3    │────→│  │ IO 线程池 │→│ 主线程      │  │
│ ...      │     │  │ (读取/写入)│ │ (命令执行)  │  │
│ 连接N    │────→│  └─────────┘  └───────────┘  │
└──────────┘     │                              │
                 │  主线程: 解析→执行→返回         │
                 │  IO线程: read()和write()       │
                 └──────────────────────────────┘

# redis.conf
io-threads 4           # IO 线程数（建议 CPU 核数的一半，不超过 8）
io-threads-do-reads yes # 开启读操作多线程（默认仅写多线程）

// src/server.h
typedef struct redisObject {
    unsigned type:4;      // 类型: OBJ_STRING/OBJ_LIST/OBJ_HASH/OBJ_SET/OBJ_ZSET
    unsigned encoding:4;  // 编码: 底层实现方式
    unsigned lru:LRU_BITS; // LRU/LFU 信息（24位）
    int refcount;          // 引用计数
    void *ptr;             // 指向底层数据结构的指针
} robj;

# 查看键的底层编码
OBJECT ENCODING mykey

// src/sds.h (Redis 7.x 简化)
struct __attribute__ ((__packed__)) sdshdr8 {
    uint8_t len;        // 已使用长度
    uint8_t alloc;      // 分配总长度（不含头和\0）
    unsigned char flags; // 头类型标记
    char buf[];          // 实际数据
};

127.0.0.1:6379> SET num 12345
OK
127.0.0.1:6379> OBJECT ENCODING num
"int"

127.0.0.1:6379> SET str "hello"
OK
127.0.0.1:6379> OBJECT ENCODING str
"embstr"

127.0.0.1:6379> SET bigstr [50个字符]
OK
127.0.0.1:6379> OBJECT ENCODING bigstr
"raw"

quicklist
├── head → quicklistNode
│             ├── *prev
│             ├── *next
│             ├── *entry (指向 listpack)
│             │     └── listpack: [element1][element2]...
│             ├── count (节点内元素数)
│             └── encoding (RAW/LZ4 压缩)
├── tail → quicklistNode
└── count (总元素数)

# 配置每个 quicklist 节点的 listpack 大小
# -1: 每个 listpack 最大 4KB
# -2: 每个 listpack 最大 8KB (默认)
# -3: 每个 listpack 最大 16KB
# -4: 每个 listpack 最大 32KB
# -5: 每个 listpack 最大 64KB
list-max-listpack-size -2

# 中间节点是否压缩 (none/all/all/小数量保留两端)
list-compress-depth 0  # 默认不压缩，1表示首尾各保留1个不压缩

# 转换阈值
hash-max-listpack-entries 128   # 元素数 <= 128 用 listpack
hash-max-listpack-value 64      # 单个值 <= 64 字节

// src/dict.h
typedef struct dict {
    dictEntry **table;      // 哈希表数组
    unsigned long size;     // 表大小（总是 2^n）
    unsigned long sizemask;  // size - 1，用于快速取模
    unsigned long used;     // 已用条目数
    // ...
} dict;

typedef struct dictEntry {
    void *key;
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    struct dictEntry *next;  // 链地址法解决冲突
} dictEntry;

typedef struct dict {
    dict ht[2];           // 两个哈希表
    int rehashidx;        // -1 表示没有 rehash
} dict;

// src/intset.h
typedef struct intset {
    uint32_t encoding;  // INTSET_ENC_INT16/32/64
    uint32_t length;
    int8_t contents[];   // 有序整数数组
} intset;

set-max-intset-entries 512  # 超过此值转为 hashtable

zset-max-listpack-entries 128
zset-max-listpack-value 64

// src/t_zset.h
typedef struct zskiplistNode {
    sds ele;                           // 元素值
    double score;                      // 分值
    struct zskiplistNode *backward;    // 后退指针
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 前进指针
        unsigned long span;            // 跨度（用于计算排名）
    } level[];                         // 多层指针数组
} zskiplistNode;

typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;              // 节点数
    int level;                         // 当前最大层数
} zskiplist;

1. 生成随机层数（ZSKIPLIST_MAXLEVEL=32, p=0.25）
   - level 1: 100% 概率
   - level 2: 25% 概率
   - level 3: 6.25% 概率
   - ...
2. 创建新节点，分配 level[] 数组
3. 从最高层开始，找到每层的插入位置（用 update[] 数组记录）
4. 在每层插入新节点，设置 forward 指针
5. 设置 backward 指针
6. 更新 span（跨度）值

// src/t_stream.h
typedef struct stream {
    rax *rax;              // Radix Tree 存储消息
    uint64_t length;       // 消息总数
    streamID maxdeleted;  // 最大已删除ID
    streamID length;       // 最后一条消息ID
    rax *nacks;            // Consumer Group 的 PEL（待确认消息）
    // ...
} stream;

# 生产消息
XADD mystream * name Alice age 30
# 消费消息
XREAD COUNT 10 STREAMS mystream 0
# 消费组
XGROUP CREATE mystream group1 0
XREADGROUP GROUP group1 consumer1 COUNT 10 STREAMS mystream >

# 手动触发
SAVE        # 同步执行，阻塞所有命令，生产禁用
BGSAVE      # fork 子进程异步执行，推荐

# 自动触发配置
save 3600 1     # 1小时内至少1个key变化
save 300 100    # 5分钟内至少100个key变化
save 60 10000   # 1分钟内至少10000个key变化

1. 主进程调用 fork()，创建子进程
2. fork 瞬间，子进程和主进程共享同一份物理内存（只读）
3. 子进程遍历所有数据，写入 RDB 文件
4. 此期间主进程仍处理命令
5. 如果主进程修改了某个页，OS 的 COW 机制会复制该页
6. 子进程写完后，通知主进程

主进程               子进程
  │                    │
  ├─ fork() ──────────→│ (共享内存)
  │                    │
  ├─ 处理命令           ├─ 遍历数据
  │  修改key A         │  写入RDB
  │  ↓                  │
  │  COW: 复制A的页     │  (读到的还是旧值)
  │                    │
  │                    ├─ RDB完成
  │←── 信号通知 ────────┤
  │                    │
  ├─ 更新状态           └─ 退出

REDIS              [5字节 magic]
[版本号]            [4字节]
[辅助字段]          [元数据：版本、时间、内存使用等]
[db_selector]      [数据库选择器]
[键值对...]        [TTL + key + value，二进制编码]
EOF                [1字节 结束标记]
[8字节校验和]       [CRC64]

appendonly yes
appendfilename "appendonly.aof"
# appendfsync always    # 每条命令都刷盘（最安全，性能最差）
appendfsync everysec   # 每秒刷盘（默认，推荐：最多丢1秒数据）
# appendfsync no        # 由OS决定刷盘（性能最好，可能丢数据较多）

命令执行
  ↓
追加到 aof_buf（内存缓冲区）
  ↓
根据 appendfsync 策略写入磁盘
  ↓
AOF 文件越来越大 → 触发 AOF Rewrite

原始AOF:
SET key1 "v1"
SET key1 "v2"    ← 已被覆盖
DEL key1         ← 已删除
LPUSH list "a"
LPUSH list "b"
LPUSH list "c"

重写后:
SET key1 "v2"    ← 不存在，跳过
LPUSH list "c" "b" "a"  ← 合并

# 自动触发配置
auto-aof-rewrite-percentage 100  # AOF 文件比上次重写后大 100%
auto-aof-rewrite-min-size 64mb   # 且 AOF 文件至少 64MB

aof-use-rdb-preamble yes  # 默认 yes

AOF 文件结构：
┌──────────────────────────┐
│ RDB 格式数据（全量快照）    │  ← 二进制，紧凑，加载快
├──────────────────────────┤
│ AOF 增量命令（重写期间的变更）│  ← 文本格式，追加
└──────────────────────────┘

fork 的开销：
1. fork 本身耗时 —— 与内存大小正相关
   - 1GB 内存: ~20ms
   - 10GB 内存: ~200ms
   - 50GB 内存: ~1s+
   fork 期间主线程阻塞，导致请求超时

2. COW 的内存开销
   - 如果 fork 期间大量写入，会复制大量内存页
   - 最坏情况：内存翻倍

3. 大量数据写入磁盘的 IO 压力

# 查看 info stats
redis-cli info stats | grep latest_fork_usec
# 单位微秒，超过 100000 (100ms) 需要关注

# 查看内存使用
MEMORY USAGE key           # 查看单个 key 占用内存
INFO memory                # 查看整体内存信息

# 关键指标
used_memory:1000000        # Redis 分配的总内存（含碎片）
used_memory_rss:1200000    # 操作系统看到的 RSS
mem_fragmentation_ratio:1.2 # RSS / used_memory
# 1.0-1.5: 正常
# > 1.5: 碎片率高
# < 1.0: 可能使用了 Swap（严重！）

// src/db.c —— 每次访问 key 前调用
int expireIfNeeded(redisDb *db, robj *key) {
    if (!keyIsExpired(db, key)) return 0;
    // 过期了，从数据库删除
    deleteExpiredKey(db, key);
    // 通知从库和AOF
    propagateDeletion(db, key);
    return 1;
}

// src/expire.c —— 定期过期
void activeExpireCycle(int type) {
    // 每次随机抽查 20 个设置了 TTL 的 key
    // 删除其中已过期的
    // 如果过期比例 > 25%，继续抽查下一批
    // 每轮执行不超过 25ms
}

// 淘汰时，随机采样 N 个 key
// maxmemory-samples 5 (默认)
// 从 N 个 key 中淘汰最近最少使用的
// 采样数越大，越接近真实 LRU，但 CPU 开销也越大

// redisObject.lru 的 24 位被拆分:
// 高 16 位: 最后访问时间（分钟精度，约 45 天循环）
// 低 8 位: 访问频率计数器（对数计数，0-255）

uint8_t lfu_log_incr(uint8_t counter) {
    if (counter == 255) return 255;
    double r = (double)rand() / RAND_MAX;
    double baseval = counter - LFU_INIT_VAL; // LFU_INIT_VAL = 5
    double p = 1.0 / (baseval * lfu_log_factor + 1.0);
    // 概率性增加，访问越多增加概率越低
    if (r < p) counter++;
    return counter;
}

# LFU 配置
maxmemory-policy allkeys-lfu
lfu-log-factor 10   # 计数器增长因子，越大增长越慢
lfu-decay-time 1     # 衰减周期（分钟）

maxmemory 8gb                    # 限制最大内存
maxmemory-policy allkeys-lru     # 缓存场景用 LRU
maxmemory-samples 10             # 采样数从 5 提高到 10，更精准

# 如果使用 LFU
# maxmemory-policy allkeys-lfu
# lfu-log-factor 10
# lfu-decay-time 1

redis-cli info memory | grep -E "used_memory_human|mem_fragmentation_ratio|maxmemory_human"

# 查看各 key 的大致内存分布
redis-cli --bigkeys          # 扫描大 key
redis-cli --memkeys          # 扫描内存占用高的 key（6.0+）

Slave                              Master
  │                                  │
  ├─ PSYNC ? -1 (第一次连接)         │
  │─────────────────────────────────→│
  │                                  ├─ 执行 BGSAVE 生成 RDB
  │                                  ├─ 生成 replication backlog
  │←── +FULLRESYNC <runid> <offset> │
  │                                  │
  │                                  ├─ 发送 RDB 文件
  │←────────── [RDB 数据] ───────────│
  │                                  │
  ├─ 加载 RDB                        ├─ 缓存期间的写命令
  │                                  │
  │←── [缓冲区命令] ─────────────────│
  │                                  │
  ├─ 执行命令，数据一致               │
  │                                  │
  │═══ 进入增量同步 ═════════════════│

Slave 断线重连
  ↓
PSYNC <runid> <offset>   # 发送上次同步的 runid 和 offset
  ↓
Master 检查:
  1. runid 是否匹配？不匹配 → 全量同步
  2. offset 是否在 backlog 中？
     - 在 → 增量同步，发送 [offset, current_offset] 的命令
     - 不在 → 全量同步（backlog 被覆盖了）

repl-backlog-size 256mb     # 环形缓冲区大小，越大越不容易全量同步
repl-backlog-ttl 3600      # backlog 空闲 TTL（无从库后多久释放）
repl-timeout 60             # 超时时间
min-replicas-to-write 1     # 至少1个从库同步成功才允许写入
min-replicas-max-lag 10    # 从库延迟不超过10秒

Master → Slave1 → Slave2 → Slave3

┌────────────┐
│ Sentinel 1 │──────┐
│ Sentinel 2 │──────┤  互相通信 + 监控 Master/Slave
│ Sentinel 3 │──────┘
└────────────┘         ┌────────┐
                      │ Master │
                      └────┬───┘
                           │
                     ┌─────┴─────┐
                     │ Slave 1   │
                     │ Slave 2   │
                     └───────────┘

1. Leader Sentinel 确认 Master 客观下线
2. 从所有 Slave 中选出最优 Slave:
   a. 过滤掉不健康的 Slave（断线时间长、响应慢）
   b. 按优先级（slave-priority）排序
   c. 优先级相同，比较复制 offset（数据最新的）
   d. offset 相同，比较 runid（最小的）
3. 对最优 Slave 执行 SLAVEOF NO ONE，升级为 Master
4. 通知其他 Slave 连接新 Master
5. 更新 Sentinel 配置，通知客户端

# sentinel.conf
sentinel monitor mymaster 192.168.1.100 6379 2  # quorum=2
sentinel down-after-milliseconds mymaster 5000   # 5秒无响应判定下线
sentinel failover-timeout mymaster 60000         # 故障转移超时
sentinel parallel-syncs mymaster 1               # 同时同步的从库数

# Key → Slot 的映射算法
slot = CRC16(key) mod 16384

Gossip 消息类型:
- PING: "我还活着，这是我知道的集群状态"
- PONG: 响应 PING
- MEET: "加入集群"
- FAIL: "XXX 节点挂了"
- PUBLISH: 广播消息（用于 Pub/Sub）

CLUSTER SETSLOT <slot> MIGRATING <target-node-id>  # 源节点：准备迁出
CLUSTER SETSLOT <slot> IMPORTING <source-node-id>  # 目标节点：准备迁入
# 迁移过程中:
# - GETKEYSINSLOT 获取槽中的 key
# - MIGRATE 逐个迁移 key
# - 迁移完成，广播槽位变更

1. 节点A PING 节点B，超过 cluster-node-timeout 未响应 → 标记 PFAIL（疑似下线）
2. 通过 Gossip 传播 PFAIL 状态
3. 超过半数主节点标记 B 为 PFAIL → 升级为 FAIL（确定下线）
4. B 的从节点检测到 B 下线 → 发起选举
5. 从节点获取半数以上主节点的投票 → 成为新主节点
6. 新主节点接管 B 的槽位，广播集群

# Hash Tag: 用 {} 指定 hash 计算的 key 部分
SET {user:1000}:name "Alice"  # slot = CRC16("user:1000") mod 16384
SET {user:1000}:age 30        # 同一个 slot，可以多键操作
MSET {user:1000}:name "Alice" {user:1000}:age 30  # ✅ 同一 slot

def get_user(user_id):
    # 1. 查缓存
    user = redis.get(f"user:{user_id}")
    if user is not None:
        if user == "NULL":
            return None  # 缓存的空值
        return json.loads(user)
    
    # 2. 查数据库
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    
    if user is None:
        # 缓存空值，设置较短 TTL
        redis.setex(f"user:{user_id}", 300, "NULL")
        return None
    
    redis.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

from pybloom_live import ScalableBloomFilter

# 启动时加载所有存在的 key 到布隆过滤器
bf = ScalableBloomFilter()
for user_id in db.query("SELECT id FROM users"):
    bf.add(str(user_id))

def get_user(user_id):
    # 1. 先过布隆过滤器
    if str(user_id) not in bf:
        return None  # 一定不存在
    
    # 2. 查缓存
    ...
    # 3. 查数据库
    ...

# 需安装 RedisBloom 模块
BF.ADD myfilter "user:1001"
BF.EXISTS myfilter "user:1001"  # 返回 1（可能存在）
BF.EXISTS myfilter "user:9999"  # 返回 0（一定不存在）

import redis
import time

r = redis.Redis()

def get_hot_data(key):
    # 1. 查缓存
    data = r.get(key)
    if data is not None:
        return json.loads(data)
    
    # 2. 缓存未命中，尝试获取锁
    lock_key = f"lock:{key}"
    if r.set(lock_key, "1", nx=True, ex=10):  # SETNX + TTL
        try:
            # 3. 获取锁成功，再次查缓存（双重检查）
            data = r.get(key)
            if data is not None:
                return json.loads(data)
            
            # 4. 查数据库
            data = db.query(key)
            
            # 5. 写缓存
            r.setex(key, 3600, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)  # 释放锁
    else:
        # 6. 获取锁失败，短暂等待后重试
        time.sleep(0.05)
        return get_hot_data(key)

def get_hot_data(key):
    data = r.get(key)
    if data is None:
        return None
    
    data_dict = json.loads(data)
    
    # 检查逻辑过期时间
    if data_dict['expire'] > time.time():
        return data_dict['data']  # 未过期，直接返回
    
    # 逻辑过期，尝试获取锁异步刷新
    if r.set(f"lock:{key}", "1", nx=True, ex=10):
        # 后台线程刷新缓存
        threading.Thread(target=refresh_cache, args=(key,))
    
    # 返回旧数据（不阻塞用户）
    return data_dict['data']

# 避免大量 key 同时过期
ttl = 3600 + random.randint(-300, 300)  # 1小时 ± 5分钟
r.setex(key, ttl, value)

请求 → 本地缓存（Caffeine/Guava）→ Redis → 数据库
         TTL: 60s                    TTL: 3600s

from circuitbreaker import circuit

@circuit(failure_threshold=10, recovery_timeout=30)
def get_data(key):
    data = redis.get(key)
    if data is None:
        data = db.query(key)  # 数据库压力大时熔断
    return data

# 熔断后返回降级数据
def get_data_safe(key):
    try:
        return get_data(key)
    except CircuitBreakerError:
        return get_fallback_data(key)  # 返回默认值或缓存

# 加锁（原子操作）
SET lock_key unique_value NX PX 30000
# NX: 不存在才设置
# PX 30000: 30秒过期

import uuid

def acquire_lock(key, ttl=30):
    value = str(uuid.uuid4())
    if redis.set(key, value, nx=True, ex=ttl):
        return value
    return None

def release_lock(key, value):
    # 必须用 Lua 脚本保证原子性
    script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    return redis.eval(script, 1, key, value)

RLock lock = redisson.getLock("myLock");
lock.lock();  // 自动续期
try {
    // 业务逻辑
} finally {
    lock.unlock();
}

// 带超时的锁
lock.lock(10, TimeUnit.SECONDS);  // 不会启动看门狗
// 尝试获取锁
boolean acquired = lock.tryLock(5, 30, TimeUnit.SECONDS);
// 5秒内尝试获取，获取到后30秒自动释放

# key: 锁名  field: 客户端ID  value: 重入次数
HSET lock_key client_id 1     # 第一次加锁
HINCRBY lock_key client_id 1  # 重入加锁
HINCRBY lock_key client_id -1 # 释放一次
# value 减为 0 时 DEL

if (redis.call('exists', KEYS[1]) == 0) then
    redis.call('hset', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
end;
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
end;
return redis.call('pttl', KEYS[1]);

1. 获取当前时间 T1
2. 依次向 N 个（通常5个）独立的 Redis 实例请求加锁
3. 每个请求设置超时时间（很短，如 50ms）
4. 如果获取了超过 N/2+1（即3个）实例的锁，且总耗时 < TTL，则认为加锁成功
5. 否则向所有实例发送释放锁请求

// Redisson RedLock 使用
RLock lock1 = redisson1.getLock("lockKey");
RLock lock2 = redisson2.getLock("lockKey");
RLock lock3 = redisson3.getLock("lockKey");

RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
redLock.lock();
try {
    // 业务逻辑
} finally {
    redLock.unlock();
}

用户 → CDN → 网关（限流）→ 应用层 → 缓存层 → 数据库

# 活动开始前预热库存到 Redis
redis.set("stock:sku:1001", 100)  # 100件库存

# 秒杀下单
def seckill(user_id, sku_id):
    # 1. 防重：SETNX 判断是否已秒杀
    if not redis.set(f"seckill:user:{user_id}:{sku_id}", "1", nx=True, ex=3600):
        return "您已秒杀成功，请勿重复参与"
    
    # 2. 扣库存（Lua 保证原子性）
    script = """
    local stock = redis.call('get', KEYS[1])
    if not stock then return -1 end
    if tonumber(stock) <= 0 then return 0 end
    redis.call('decr', KEYS[1])
    return 1
    """
    result = redis.eval(script, 1, f"stock:sku:{sku_id}")
    
    if result == 1:
        # 3. 发送 MQ 异步创建订单
        mq.send("order_queue", {"user_id": user_id, "sku_id": sku_id})
        return "秒杀成功，订单创建中"
    elif result == 0:
        return "已售罄"
    else:
        return "活动未开始"

# Lua 实现令牌桶限流
LUA_TOKEN_BUCKET = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])  -- 桶容量
local rate = tonumber(ARGV[2])       -- 令牌生成速率（个/秒）
local now = tonumber(ARGV[3])        -- 当前时间戳（毫秒）
local requested = tonumber(ARGV[4])  -- 请求的令牌数

local bucket = redis.call('hmget', key, 'tokens', 'timestamp')
local tokens = tonumber(bucket[1]) or capacity
local last_time = tonumber(bucket[2]) or now

-- 计算时间差内生成的令牌
local delta = math.max(0, now - last_time) / 1000
tokens = math.min(capacity, tokens + delta * rate)

if tokens < requested then
    return 0  -- 令牌不足
end

tokens = tokens - requested
redis.call('hmset', key, 'tokens', tokens, 'timestamp', now)
redis.call('expire', key, math.ceil(capacity / rate) + 1)
return 1
"""

def allow_request(key, capacity=100, rate=10, requested=1):
    now = int(time.time() * 1000)
    return bool(redis.eval(LUA_TOKEN_BUCKET, 1, key, capacity, rate, now, requested))

# 秒杀成功排行榜
redis.zadd("seckill:rank:sku:1001", {user_id: timestamp})

# 获取前 10 名
redis.zrange("seckill:rank:sku:1001", 0, 9, withscores=True)

# 游戏积分排行榜
redis.zadd("game:rank", {user_id: score})

# 获取Top10
redis.zrevrange("game:rank", 0, 9, withscores=True)

# 获取用户排名
redis.zrevrank("game:rank", user_id)  # 返回排名（0开始）
redis.zscore("game:rank", user_id)     # 返回分数

# 获取排名区间（如 90-100名）
redis.zrevrange("game:rank", 89, 99, withscores=True)

# 分页查询
page, size = 1, 20
redis.zrevrange("game:rank", (page-1)*size, page*size-1, withscores=True)

# 百万级用户排行榜
# 1. 定时刷新（每5分钟全量更新）
# 2. 只缓存 Top 1000，之外的用数据库
# 3. 用户查询自己的排名用 ZREVRANK（O(logN)）

# 分段缓存优化（避免单个大 ZSet）
def get_rank(user_id):
    score = redis.zscore("game:rank", user_id)
    if score is None:
        return None
    
    # Top 1000 实时查询
    if score > redis.zscore("game:rank:top1000",
                            redis.zrevrange("game:rank:top1000", -1, -1)[0]):
        return redis.zrevrank("game:rank:top1000", user_id) + 1
    
    # 其余分段查询
    segment = int(score / 1000)  # 按 1000 分分段
    return redis.zrevrank(f"game:rank:seg:{segment}", user_id) + 1 + offset

# 每分钟限制 100 次
def rate_limit_fixed(user_id, limit=100):
    key = f"rate:{user_id}:{int(time.time() / 60)}"
    count = redis.incr(key)
    if count == 1:
        redis.expire(key, 60)
    return count <= limit

# 使用 ZSet 实现
def rate_limit_sliding(user_id, limit=100, window=60):
    key = f"rate:{user_id}"
    now = time.time()
    pipe = redis.pipeline()
    
    # 1. 移除窗口外的记录
    pipe.zremrangebyscore(key, 0, now - window)
    # 2. 添加当前请求
    pipe.zadd(key, {str(uuid.uuid4()): now})
    # 3. 统计窗口内请求数
    pipe.zcard(key)
    # 4. 设置过期时间
    pipe.expire(key, window)
    
    results = pipe.execute()
    return results[2] <= limit

# 生产者
redis.lpush("task_queue", json.dumps(task))

# 消费者（阻塞式）
while True:
    result = redis.brpop("task_queue", timeout=30)
    if result:
        task = json.loads(result[1])
        process(task)

# 生产者
redis.xadd("mystream", {"field1": "value1", "field2": "value2"})

# 消费者（消费者组）
# 创建消费者组
redis.xgroup_create("mystream", "group1", id="0", mkstream=True)

# 消费消息
while True:
    messages = redis.xreadgroup(
        groupname="group1",
        consumername="consumer1",
        streams={"mystream": ">"},
        count=10,
        block=5000
    )
    for stream, msg_list in messages:
        for msg_id, fields in msg_list:
            process(fields)
            redis.xack("mystream", "group1", msg_id)  # 确认消息

# 死信队列：处理 pending 时间过长的消息
pending = redis.xpending("mystream", "group1")
if pending['pending'] > 0:
    # 查看待确认消息
    messages = redis.xpending_range("mystream", "group1", min="-", max="+", count=10)
    for msg in messages:
        if msg['time_since_delivered'] > 60000:  # 超过60秒
            # 转移给其他消费者
            redis.xclaim("mystream", "group1", "consumer2",
                        min_idle_time=60000, message_ids=[msg['message_id']])

# 添加延迟任务
def add_delay_task(task_id, task_data, delay_seconds):
    execute_time = time.time() + delay_seconds
    redis.zadd("delay_queue", {json.dumps({"id": task_id, "data": task_data}): execute_time})

# 消费者轮询
def consume_delay_queue():
    while True:
        # 获取到期任务
        now = time.time()
        tasks = redis.zrangebyscore("delay_queue", 0, now, start=0, num=10)
        for task_json in tasks:
            # ZREM 返回 1 表示成功抢到（防止重复消费）
            if redis.zrem("delay_queue", task_json):
                task = json.loads(task_json)
                process(task)
        time.sleep(0.1)

# 开启键空间通知
redis.config_set("notify-keyspace-events", "Ex")

# 添加延迟任务（设置过期时间 = 延迟时间）
redis.setex(f"delay:task:{task_id}", delay_seconds, task_data)

# 监听过期事件
pubsub = redis.pubsub()
pubsub.psubscribe("__keyevent@0__:expired")

for message in pubsub.listen():
    if message['type'] == 'pmessage':
        key = message['data']  # delay:task:xxx
        task_id = key.split(':')[-1]
        task_data = get_task_data(task_id)
        process(task_data)

# 方法1: redis-cli --bigkeys
redis-cli --bigkeys

# 方法2: 手动扫描
redis-cli --eval scan_bigkeys.lua

# 方法3: MEMORY USAGE
redis-cli memory usage key

# 1. 拆分大 Hash
# 原: user:1000 (包含 name, age, email, address, ...)
# 拆: user:1000:basic (name, age)
#     user:1000:contact (email, phone)
#     user:1000:address (...)

# 2. 拆分大 List
# 原: feed:user:1000 (10万条)
# 拆: feed:user:1000:page:0  (1000条)
#     feed:user:1000:page:1  (1000条)

# 3. 异步删除大 Key（Redis 4.0+）
# UNLINK 替代 DEL（后台线程异步释放内存）
redis.unlink("big_key")

# 配置自动异步删除
# lazyfree-lazy-eviction yes      # 淘汰时异步
# lazyfree-lazy-expire yes        # 过期时异步
# lazyfree-lazy-server-del yes    # 服务器删除时异步
# lazyfree-lazy-user-del yes     # 用户 DEL 时异步

# 方法1: redis-cli --hotkeys (需开启 LFU)
redis-cli --hotkeys

# 方法2: MONITOR 命令（生产慎用，影响性能）

# 方法3: 代理层统计（如果有代理层）

# 1. 本地缓存（多级缓存）
def get_hot_data(key):
    # 先查本地缓存
    data = local_cache.get(key)  # Caffeine / Guava
    if data:
        return data
    
    # 再查 Redis
    data = redis.get(key)
    if data:
        local_cache.set(key, data, ttl=10)  # 本地缓存10秒
    return data

# 2. 拆分热 Key（读写分离到多个副本）
# 原单节点热 key: hot:config
# 拆: hot:config:1, hot:config:2, hot:config:3
# 客户端随机访问其中一个
def get_hot_data(key):
    n = 3  # 副本数
    idx = random.randint(1, n)
    return redis.get(f"{key}:{idx}")

# 3. Cluster 下使用读写分离
# 热点 key 的读请求分散到从节点

# 逐条操作（10次网络往返）
for i in range(10):
    redis.set(f"key{i}", f"value{i}")

# Pipeline（1次网络往返）
pipe = redis.pipeline()
for i in range(10):
    pipe.set(f"key{i}", f"value{i}")
results = pipe.execute()

# MULTI/EXEC 事务
MULTI
SET key1 "value1"
INCR counter
SET key2 "value2"
EXEC

# 注意: Redis 事务不支持回滚！
# 如果 INCR 失败（counter 不是数字），后续命令仍会执行

# Lua 脚本（原子执行，推荐）
script = """
local current = redis.call('get', KEYS[1])
if current == false then
    redis.call('set', KEYS[1], ARGV[1])
    redis.call('expire', KEYS[1], ARGV[2])
    return 1
else
    return 0
end
"""
# EVAL 执行，Redis 保证原子性
redis.eval(script, 1, "key", "value", 60)

# Lua 脚本超时配置
lua-time-limit 5000  # 毫秒
# 超时后可以 SCRIPT KILL 中止（但可能只执行了一半）
# 如果已经执行了写操作，只能 SHUTDOWN NOSAVE

# ===== 内存 =====
maxmemory 8gb
maxmemory-policy allkeys-lru
maxmemory-samples 10

# ===== 持久化 =====
appendonly yes
appendfsync everysec
aof-use-rdb-preamble yes
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb
save 3600 1
save 300 100
save 60 10000

# ===== 网络 =====
timeout 300                    # 客户端空闲超时
tcp-keepalive 60               # TCP keepalive
tcp-backlog 511                # 连接队列
protected-mode yes             # 保护模式

# ===== 复制 =====
repl-backlog-size 256mb
repl-timeout 60
min-replicas-to-write 0        # 生产可设 1
min-replicas-max-lag 10

# ===== 性能 =====
hz 10                          # 后台任务频率
activerehashing yes            # 渐进式 rehash
io-threads 4                  # IO 多线程（6.0+）
lazyfree-lazy-eviction yes    # 异步释放内存
lazyfree-lazy-expire yes
lazyfree-lazy-server-del yes
lazyfree-lazy-user-del yes

# ===== 客户端缓冲区 =====
client-output-buffer-limit normal 0 0 0
client-output-buffer-limit replica 256mb 64mb 60
client-output-buffer-limit pubsub 32mb 8mb 60

# Redis 完整监控工具
# 1. redis-stat（实时监控）
redis-stat --server

# 2. RedisLive（Web 可视化）
# 3. Prometheus + redis_exporter + Grafana（推荐）
# 4. redis-cli --latency（延迟监控）

# 1. 检查慢查询
redis-cli slowlog get 10

# 2. 检查是否有大 Key 操作
redis-cli --bigkeys

# 3. 检查 Lua 脚本
redis-cli script exists <sha1>

# 4. 检查 MONITOR 是否开启（极度影响性能）
redis-cli info clients

# 1. 检查是否有大 Key
redis-cli --bigkeys

# 2. 检查过期策略
redis-cli config get maxmemory-policy

# 3. 检查客户端输出缓冲区
redis-cli client list
# 关注 omem（输出缓冲区内存）

# 4. 检查碎片率
redis-cli info memory | grep fragmentation

# 5. 手动整理碎片（4.0+）
redis-cli memory purge

# 1. 查看客户端列表
redis-cli client list
# 关注 age（连接时长）和 cmd（最后执行的命令）

# 2. 检查是否有空闲连接
redis-cli client list | grep -v "cmd=.*"

# 3. 设置超时
redis-cli config set timeout 300

# 4. 检查连接池配置
# 应用侧 HikariCP/Druid 的 maxPoolSize 是否合理

# 事务示例（不支持条件逻辑）
MULTI
GET balance
# 这里不能根据 GET 的结果做判断
SET balance 100
EXEC

# Lua 脚本示例（支持条件逻辑）
EVAL "local bal = redis.call('get', KEYS[1]) \
      if tonumber(bal) >= tonumber(ARGV[1]) then \
          return redis.call('decrby', KEYS[1], ARGV[1]) \
      else return -1 end" 1 balance 50