# 启动 Redis
redis-server

# 客户端连接
redis-cli

# 基本命令
SET name "Alice"        # 设置字符串
GET name                # 获取字符串
HSET user:1 age 25      # 设置 Hash 字段
HGET user:1 age         # 获取 Hash 字段
LPUSH tasks "task1"     # List 左推入
RPOP tasks              # List 右弹出
SADD tags "java"        # Set 添加成员
ZADD rank 100 "alice"   # ZSet 添加成员

import redis.clients.jedis.Jedis;

Jedis jedis = new Jedis("127.0.0.1", 6379);
jedis.set("greeting", "Hello Redis");
String value = jedis.get("greeting");
jedis.close();

# Redis 配置示例（redis.conf）
bind 0.0.0.0
port 6379
# 开启 AOF 持久化
appendonly yes
appendfilename "appendonly.aof"
# 配置内存上限
maxmemory 4gb
# 配置淘汰策略
maxmemory-policy allkeys-lru
# I/O 多线程（Redis 6+）
io-threads 4
io-threads-do-reads yes

# 查看 Redis 实时性能指标
redis-cli INFO stats
redis-cli INFO commandstats

# 查看 I/O 多线程配置
redis-cli CONFIG GET io-threads
redis-cli CONFIG GET io-threads-do-reads

# 性能测试
redis-cli --benchmark -n 100000 -c 50 -d 1024

主线程（事件循环）:
  1. 接收客户端连接 → 注册可读事件
  2. 分发 I/O 读任务到 I/O 线程池
  3. 从输入缓冲区读取命令 → 命令队列

I/O 线程池（读线程）:
  1. 从 socket 读取数据到输入缓冲区

命令执行线程（主线程）:
  1. 从命令队列取出命令
  2. 执行内存操作
  3. 结果写入输出缓冲区

I/O 线程池（写线程）:
  1. 从输出缓冲区写数据到 socket

# 启用 I/O 多线程
io-threads 4        # I/O 线程数量，建议 = CPU 核数
io-threads-do-reads yes  # 读操作也用多线程

# 检查当前线程模型
redis-cli CONFIG GET io-threads
# 输出: 1 表示单线程 I/O；>1 表示多线程 I/O

# 动态调整 I/O 线程数
redis-cli CONFIG SET io-threads 4

# 查看命令执行耗时（验证命令阻塞）
redis-cli SLOWLOG GET 10

CPU（多核）
├── Core 0
│   └── 可运行 1-2 个线程（超线程）
├── Core 1
│   └── 可运行 1-2 个线程
├── ...
└── Core N
    └── 可运行 1-2 个线程

操作系统调度：
- 进程 ← 资源分配基本单位
- 线程 ← CPU 调度基本单位
- 一个进程内可包含多个线程

# 查看 CPU 核数（Linux）
nproc
# 或
lscpu | grep "^CPU(s)"

# 查看 Redis 线程数
redis-cli INFO server | grep thread

# 查看 Redis 各线程 CPU 占用
top -H -p $(pgrep redis-server)

// 原子自增
Long views = jedis.incr("article:views:" + articleId);
// 设置初始过期时间（首次计数时）
if (views == 1) {
    jedis.expire("article:views:" + articleId, 3600);
}

// 添加玩家分数
jedis.zadd("game:rank", 100, "player:1");
jedis.zadd("game:rank", 95, "player:2");
// 获取 Top 10
Set<String> top10 = jedis.zrevrange("game:rank", 0, 9);
// 获取玩家排名
Long rank = jedis.zrevrank("game:rank", "player:1");

// 生产者
jedis.lpush("queue:orders", orderJson);
// 消费者（阻塞等待）
List<String> result = jedis.brpop(0, "queue:orders");

// Lua 脚本实现滑动窗口
String script =
"local key = KEYS[1] " +
"local now = tonumber(ARGV[1]) " +
"local window = tonumber(ARGV[2]) " +
"local limit = tonumber(ARGV[3]) " +
"redis.call('ZREMRANGEBYSCORE', key, 0, now - window) " +
"local count = redis.call('ZCARD', key) " +
"if count < limit then " +
"  redis.call('ZADD', key, now, now .. '-' .. math.random()) " +
"  redis.call('EXPIRE', key, window / 1000) " +
"  return 1 " +
"else return 0 end";

值为整数且在 long 范围内 → int 编码（直接存 long 值）
  ↓
值为 ≤ 44 字节的字符串 → embstr 编码（embedded string，SDS header + 数据连续存储）
  ↓
值为 > 44 字节的字符串 → raw 编码（SDS 在堆上分配）

元素 ≤ 128 且所有值 ≤ 64 字节 → ziplist（紧凑双向链表）
  ↓ 超过阈值
→ hashtable（两个哈希表，渐进式 rehash）

元素 ≤ 128 且所有值 ≤ 64 字节 → ziplist（Redis 3.2+ 用 quicklist）
  ↓ 超过阈值
→ quicklist（链表 + ziplist 节点）

全为整数且 ≤ 512 元素 → intset（有序整数数组，二分查找）
  ↓ 超过阈值或非整数
→ hashtable

元素 ≤ 128 且所有值 ≤ 64 字节 → ziplist
  ↓ 超过阈值
→ skiplist + hashtable（跳表用于范围查询，哈希表用于 O(1) 查找）

# 查看 Key 的编码类型
redis-cli OBJECT ENCODING key_name

# 示例
SET int_key 123        # 返回: "int"
SET str_key "hello"   # 返回: "embstr"
SET long_key "very long string..."  # 返回: "raw"

# 查看 Hash 编码
HSET small_hash f1 v1  # 返回: "ziplist"（元素少时）

// 查看对象编码
String encoding = jedis.objectEncoding("myKey");
System.out.println(encoding); // "int", "embstr", "raw", "ziplist", "hashtable" 等

hash-max-listpack-entries 128    # Hash 元素阈值（Redis 7+）
hash-max-listpack-value 64        # Hash 值长度阈值
list-max-listpack-size -2         # List 节点大小阈值
set-max-intset-entries 512        # Set 整数集合阈值
zset-max-listpack-entries 128     # ZSet 元素阈值
zset-max-listpack-value 64        # ZSet 值长度阈值

typedef struct {
    int len;      // 字符串长度
    int free;     // 已分配但未使用的空间
    char buf[];   // 实际存储数据的柔性数组
} sdshdr;

embstr 编码（len ≤ 44 字节）：
┌──────────┬──────────┬──────────────────────┐
│ sdshdr   │ len      │ buf[] (数据)         │
└──────────┴──────────┴──────────────────────┘
  连续存储，一次 malloc

raw 编码（len > 44 字节）：
┌──────────┬──────────┬──────────────────────┐
│ sdshdr*  │ len      │ buf[] (数据)         │
└──────────┴──────────┴──────────────────────┘
  sdshdr 和 buf 分离，两次 malloc

1. 值为整数（long 范围内） → int 编码
   例: SET num 123 → int 编码，直接存 long

2. 值长度 ≤ 44 字节 → embstr 编码
   例: SET name "Alice" → embstr 编码

3. 值长度 > 44 字节 → raw 编码
   例: SET desc "A very long..." → raw 编码

4. 对 embstr 执行修改操作 → 自动转 raw 编码
   例: APPEND name "extra" → raw 编码

修改后所需长度 < 预分配空间 → 不 realloc
修改后所需长度 < 1MB → 扩容为所需长度 × 2（预分配翻倍）
修改后所需长度 ≥ 1MB → 扩容为所需长度 + 1MB（预分配 1MB）

# 设置字符串
SET key "value"
SET counter 100

# 整数自增（原子操作）
INCR counter       # → 101
INCRBY counter 10  # → 111
DECR counter       # → 110

# 字符串操作
APPEND key "_suffix"   # 追加
GETRANGE key 0 4        # 获取子串
SETRANGE key 5 "world"  # 替换子串
STRLEN key              # 获取长度

# 查看编码
OBJECT ENCODING key

// 使用 Jedis
jedis.set("greeting", "Hello Redis");
jedis.incr("counter");
String encoding = jedis.objectEncoding("greeting"); // "embstr"

// Pipeline 批量操作
Pipeline pipe = jedis.pipelined();
pipe.set("k1", "v1");
pipe.set("k2", "v2");
pipe.sync();

HSET user:1 name "Alice" age 25 gender "F"
→ Key: user:1
  Value: {name: "Alice", age: "25", gender: "F"}

┌──────┬──────┬──────┬──────┬──────┬──────┐
│key1  │val1  │key2  │val2  │...   │Zlen  │
└──────┴──────┴──────┴──────┴──────┴──────┘
  按 key-value 对顺序存储，双向遍历
  查找需遍历，O(N) 但 N ≤ 128，很快

rehash 前:
  table[0] → [bucket0] → entry1 → entry2 → NULL
  table[1] → [bucket1] → NULL
  ...
  table[N] → [bucketN] → entry3 → NULL

rehash 中（两个表）:
  old_table: → 正在迁移的旧表
  new_table: → 接收新元素的新表
  rehash_idx: 当前迁移进度

# Hash 基本操作
HSET user:1 name "Alice" age 25      # 设置字段
HGET user:1 name                      # 获取字段
HGETALL user:1                        # 获取所有字段
HMSET user:2 name "Bob" age 30       # 批量设置
HDEL user:1 age                       # 删除字段
HLEN user:1                           # 获取字段数
HEXISTS user:1 name                   # 判断字段是否存在

# 查看编码
OBJECT ENCODING user:1   # "ziplist" 或 "hashtable"

# 扫描 Hash 字段（避免阻塞）
HSCAN user:1 0 COUNT 100

Map<String, String> userMap = new HashMap<>();
userMap.put("name", "Alice");
userMap.put("age", "25");
jedis.hset("user:1", userMap);

String name = jedis.hget("user:1", "name");
Map<String, String> result = jedis.hgetAll("user:1");

hash-max-listpack-entries 128    # 最大元素数
hash-max-listpack-value 64        # 单值最大字节数

quicklist → [node1] → [node2] → ... → [nodeN]
              ↓         ↓                  ↓
           ziplist    ziplist            ziplist
           (紧凑存储)  (紧凑存储)         (紧凑存储)

intset → [1, 5, 10, 25, 100, ...]
          有序整数数组，二分查找 O(log N)
          不支持非整数元素

skiplist（跳表）:
  head → [level 3: null]
       → [level 2: 5, 25, null]
       → [level 1: 1, 5, 10, 25, 100, null]
       → [level 0: 1→node1, 5→node2, 10→node3, ...]  ← 实际节点

hashtable:
  "member1" → node1（O(1) 查找）
  "member2" → node2
  ...

# List 操作
LPUSH queue "task1"       # 左推入
RPUSH queue "task2"       # 右推入
LPOP queue                # 左弹出
RPOP queue                # 右弹出
BLPOP queue 30            # 阻塞左弹出（超时 30s）
LRANGE queue 0 -1         # 获取所有元素
LINDEX queue 2            # 按索引访问

# Set 操作
SADD tags "java" "redis"  # 添加元素
SMEMBERS tags             # 获取所有元素
SISMEMBER tags "java"     # 判断是否存在
SINTER set1 set2          # 交集
SUNION set1 set2          # 并集
SDIFF set1 set2           # 差集
SRANDMEMBER tags 3        # 随机获取 3 个元素

# ZSet 操作
ZADD rank 100 "alice"     # 添加元素（分数 100）
ZADD rank 95 "bob"
ZSCORE rank "alice"       # 获取分数
ZRANK rank "alice"        # 获取排名（升序）
ZREVRANK rank "alice"     # 获取排名（降序）
ZRANGE rank 0 9           # 获取 Top 10（升序）
ZREVRANGE rank 0 9        # 获取 Top 10（降序）
ZRANGEBYSCORE rank 90 100 # 按分数范围查询

Level 3:  1 ────────────────────────── 50
          |                             |
Level 2:  1 ────────────── 25 ──────── 50
          |                 |          |
Level 1:  1 ──── 10 ────── 25 ── 35 ── 50
          |        |        |      |    |
Level 0:  1─5─10─15─20─25─30─35─40─45─50
          (实际数据层，有序链表)

typedef struct zskiplistNode {
    sds ele;                        // 成员对象
    double score;                   // 分数
    struct zskiplistNode *backward; // 后退指针（仅最底层使用）
    struct zskiplistLevel {
        struct zskiplistNode *forward;  // 前进指针
        unsigned long span;            // 跨度（到下一个节点之间跳过了多少节点）
    } level[];                       // 层数组，柔性数组
} zskiplistNode;

# ZSet 命令底层使用跳表
ZADD rank 100 "alice"
ZADD rank 95 "bob"
ZADD rank 105 "charlie"

# 范围查询（跳表的强项）
ZRANGEBYSCORE rank 90 110    # 按分数范围
ZRANK rank "alice"           # 按成员查找排名
ZREVRANGE rank 0 9           # 逆序范围查询

public class SkipList<K extends Comparable<K>, V> {
    private static final int MAX_LEVEL = 32;
    private static final double P = 0.25;

    private class Node {
        K key;
        V value;
        Node[] forward;
        int[] span;

        Node(int level) {
            forward = new Node[level + 1];
            span = new int[level + 1];
        }
    }

    private Node head = new Node(MAX_LEVEL);
    private int level = 0;
    private int size = 0;

    private int randomLevel() {
        int lvl = 0;
        while (Math.random() < P && lvl < MAX_LEVEL) {
            lvl++;
        }
        return lvl;
    }

    public V get(K key) {
        Node current = head;
        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null
                   && current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
            }
        }
        current = current.forward[0];
        if (current != null && current.key.compareTo(key) == 0) {
            return current.value;
        }
        return null;
    }

    public void put(K key, V value) {
        Node[] update = new Node[MAX_LEVEL + 1];
        int[] rank = new int[MAX_LEVEL + 1];
        Node current = head;

        for (int i = level; i >= 0; i--) {
            rank[i] = (i == level) ? 0 : rank[i + 1];
            while (current.forward[i] != null
                   && current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
                rank[i] += current.span[i];
            }
            update[i] = current;
        }

        int newLevel = randomLevel();
        if (newLevel > level) {
            for (int i = level + 1; i <= newLevel; i++) {
                update[i] = head;
                rank[i] = 0;
                head.span[i] = size;
            }
            level = newLevel;
        }

        Node newNode = new Node(newLevel);
        newNode.key = key;
        newNode.value = value;

        for (int i = 0; i <= newLevel; i++) {
            newNode.forward[i] = update[i].forward[i];
            update[i].forward[i] = newNode;

            int span = rank[0] + 1 - rank[i];
            newNode.span[i] = update[i].span[i] - span;
            update[i].span[i] = span;
        }

        for (int i = newLevel + 1; i <= level; i++) {
            update[i].span[0]++;
        }

        size++;
    }
}

读请求 → Redis GET → 命中? → 返回（Cache Hit）
                      ↓ 未命中
                   查询 DB → 写入 Redis → 返回（Cache Miss）

用户登录 → HSET user:{id} name "xxx" age 25 email "xxx"
读用户 → HGETALL user:{id}
更新属性 → HSET user:{id} age 26（部分更新，不影响其他字段）

生产者 → LPUSH order:queue "订单JSON"
消费者 → BRPOP order:queue 0（阻塞等待）
       → 处理订单 → 自动移除

用户访问 → SADD uv:page1 "user:1" → 返回 1（新用户）
         → SADD uv:page1 "user:1" → 返回 0（已存在）

玩家得分 → ZADD rank 100 "player:1"
查排名 → ZREVRANK rank "player:1"
取 Top10 → ZREVRANGE rank 0 9

// 签到（每一天对应一个 bit）
jedis.setbit("sign:uid:1", dayOfYear, true);
// 查询某月签到天数
long signCount = jedis.bitcount("sign:uid:1");

// 记录访客
jedis.pfadd("uv:article:1", "user:1", "user:2");
// 获取 UV 近似值
long uv = jedis.pfcount("uv:article:1");

// 生产者
jedis.xadd("stream:orders",
    StreamAddParams.map("order", orderJson));
// 消费者（消费组模式）
List<StreamEntry> messages = jedis.xreadGroup(
    "group1", "stream:orders",
    new HashMap<String, StreamEntryID>() {{
        put("stream:orders", StreamEntryID.last());
    }}, 10);

// 计算共同好友
Set<String> commonFriends = jedis.sinter("user:1:friends", "user:2:friends");

触发条件 → 执行 BGSAVE
  1. Redis 进程 fork() 子进程
  2. 子进程将内存数据写入临时 .rdb 文件
  3. 写入完成，替换旧的 dump.rdb
  4. 子进程退出

触发方式：
  - save 命令：同步阻塞（不推荐）
  - bgsave 命令：fork 子进程，不阻塞主线程
  - 自动触发：满足 save 规则时自动 bgsave

[magic "REDIS"] [version] [database info] [key-value pairs] [checksum]

写命令执行 → 追加到 AOF 缓冲区
  → 根据策略刷到磁盘：
     - always：每条命令 fsync，最安全，最慢
     - everysec：每秒 fsync（默认），折中方案
     - no：交给 OS 决定，最快，最不安全

触发重写 → bgrewriteaof
  1. fork() 子进程
  2. 子进程扫描当前内存数据，写入新 AOF 文件
  3. 重写期间新命令写入 AOF 重写缓冲区
  4. 子进程完成后，缓冲区内容追加到新文件
  5. 替换旧 AOF 文件

bgrewriteaof 触发
  → 前半部分：以 RDB 格式快速写入（压缩、高效）
  → 后半部分：以 AOF 格式写入增量命令
  → 重启恢复时：先加载 RDB 部分（快速），再回放 AOF 尾部

# RDB 配置
save 900 1      # 900秒内至少1次修改触发bgsave
save 300 10     # 300秒内至少10次修改
save 60 10000   # 60秒内至少10000次修改
stop-writes-on-bgsave-error yes  # bgsave失败时停止写入
rdbcompression yes  # 压缩RDB文件
rdbchecksum yes     # 校验和
dbfilename dump.rdb
dir /var/lib/redis

# AOF 配置
appendonly yes
appendfilename "appendonly.aof"
appendfsync everysec  # always | everysec | no
auto-aof-rewrite-percentage 100  # AOF文件比上次重写增长100%时触发
auto-aof-rewrite-min-size 64mb

# 混合持久化配置
aof-use-rdb-preamble yes  # 开启混合持久化

# 手动触发
redis-cli BGSAVE       # 手动触发RDB快照
redis-cli BGREWRITEAOF # 手动触发AOF重写
redis-cli LASTSAVE     # 查看上次保存时间

启动 → 检查是否有 dump.rdb
  → fork 加载进程 → 读取 RDB 文件 → 反序列化 → 恢复数据 → 接受连接

启动 → 检查是否有 appendonly.aof
  → 读取 AOF 文件 → 逐条回放命令 → 恢复数据 → 接受连接

# 同时开启 RDB 和 AOF
# RDB 配置（可选）
save 900 1
save 300 10

# AOF 配置（推荐）
appendonly yes
appendfsync everysec

# 混合持久化（Redis 4.0+，推荐）
aof-use-rdb-preamble yes

# 查看持久化状态
redis-cli INFO persistence

时间线：
  T0: BGSAVE 执行完成，生成 dump.rdb
  T1: Redis 接收写命令 W1、W2、W3
  T2: Redis 接收写命令 W4、W5
  T3: Redis 进程异常退出（崩溃）

  结果：W1-W5 数据丢失，因为它们在 T0 之后、下一次 BGSAVE 之前

save 900 1       # 900秒内1次修改触发 → 可能丢 900 秒
save 300 10      # 300秒内10次修改触发 → 可能丢 300 秒
save 60 10000    # 60秒内10000次修改触发 → 可能丢 60 秒

注意：三个条件是"或"的关系，任一满足即触发

# 配置更短的保存间隔以减少丢失
save 60 10000
save 30 100000

# 或关闭 RDB 自动保存，手动控制
# save ""

# 配合 AOF 使用作为双保险
appendonly yes
appendfsync everysec

# 正常关闭（触发保存）
redis-cli SHUTDOWN

# 异常恢复后的检查
redis-cli LASTSAVE  # 查看最后保存时间

1. 触发条件：
   - 手动执行 BGREWRITEAOF
   - 自动触发：AOF 文件比上次重写后增长超过 auto-aof-rewrite-percentage

2. Redis 主进程：
   a. 检查是否已有重写在进行（避免并发重写）
   b. fork() 子进程
   c. 子进程开始重写

3. 子进程（重写执行）：
   a. 遍历所有数据库
   b. 对每个 Key 生成一条最终写入命令
   c. 写入临时 AOF 文件（new_appendonly.aof）
   d. 完成后通知主进程

4. 主进程（重写期间）：
   a. 正常处理命令
   b. 将新的写命令同时写入：
      - 旧 AOF 文件（保证持久化）
      - AOF 重写缓冲区（用于追加到新文件）

5. 重写完成：
   a. 主进程将 AOF 重写缓冲区内容追加到新 AOF 文件
   b. 替换旧 AOF 文件
   c. 释放缓冲区

重写时：
  → 前半部分：以 RDB 格式快速写入
  → 后半部分：以 AOF 格式写入增量命令
  → 生成混合格式文件

# 手动触发 AOF 重写
redis-cli BGREWRITEAOF

# 配置自动重写
auto-aof-rewrite-percentage 100   # 文件增长 100% 时触发
auto-aof-rewrite-min-size 64mb    # 最小重写文件大小

# 查看重写状态
redis-cli INFO persistence
# 关注：aof_rewrite_in_progress, aof_last_rewrite_time_sec

# 配置重写期间的 fsync 行为
aof-rewrite-incremental-fsync yes

jedis.bgrewriteAof();
// 或使用异步监听
jedis.bgsave(); // RDB 快照类似

混合 AOF 文件结构：
┌──────────────────────────┬────────────────────────────┐
│  RDB 格式（二进制）       │  AOF 格式（增量命令）       │
│  （某一时刻的完整数据快照） │  （快照后的增量写命令）     │
└──────────────────────────┴────────────────────────────┘
  ↑ 恢复时先加载这部分（快速）    ↑ 再回放这部分（完整）

1. appendonly = yes（AOF 已开启）
2. aof-use-rdb-preamble = yes

1. 触发 BGREWRITEAOF
2. fork 子进程
3. 子进程遍历内存数据
   → 前 80% 数据：以 RDB 二进制格式写入新文件
   → 后 20%：以 AOF RESP 格式写入增量命令
4. 重写期间的新命令写入 AOF 重写缓冲区
5. 缓冲区内容以 AOF 格式追加到文件末尾
6. 替换旧文件

1. 检查 AOF 文件头部标识
   → 如果是混合格式（RDB 前缀 + AOF 后缀）
2. 加载 RDB 部分（快速，二进制直接反序列化）
3. 回放 AOF 尾部（增量命令重放）
4. 恢复完成

# 开启混合持久化
# 必须先开启 AOF
appendonly yes
aof-use-rdb-preamble yes

# 触发混合格式重写
redis-cli BGREWRITEAOF

# 查看当前持久化状态
redis-cli INFO persistence
# 关注：aof_enabled, aof_rewrite_in_progress

# 重启 Redis 后，查看加载信息
redis-cli INFO server | grep process_id
# 检查 AOF 文件格式
xxd appendonly.aof | head
# RDB 部分以 "REDIS" magic number 开头

save 3600 1      # 宽松的保存策略
save 300 100
stop-writes-on-bgsave-error yes
rdbcompression yes

appendonly yes
appendfsync everysec
aof-use-rdb-preamble yes
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

appendonly yes
appendfsync everysec
# RDB 作为备份
save 900 1
save 300 10

appendonly yes
appendfsync always

客户端 GET key
  → 检查 expires 字典中 key 的过期时间
  → 如果 now > expire_time：
      a. 从数据库字典删除 key
      b. 从 expires 字典删除 key
      c. 通知 AOF/从服务器（传播删除事件）
      d. 返回 nil（key 不存在）
  → 如果未过期或未设置过期：
      返回值

Redis 周期性执行（每 100ms 左右）：
  1. 从设置了过期时间的 Key 中随机抽取 20 个
  2. 检查每个 Key 是否过期
  3. 删除已过期的 Key
  4. 如果本轮删除比例 > 25%：
     → 立即重复步骤 1-3（最多 20 次）
  5. 最多耗时 25% 的 CPU 时间，避免阻塞主线程

Redis 使用两个字典存储 Key：
  - db[i].dict：存储所有 Key → Value
  - db[i].expires：存储设置了过期时间的 Key → 过期时间戳

过期检查：
  → 检查 key 是否在 expires 字典中
  → 如果存在，比较当前时间与过期时间
  → 过期则删除

expires 字典的 Value 是过期时间戳（毫秒级）：
  SET key value PX 3600000
  → expires[key] = 当前时间 + 3600000（毫秒）

# 设置过期时间
SET key value EX 3600      # 3600秒后过期
SET key value PX 3600000   # 3600000毫秒后过期
EXPIRE key 60              # 60秒后过期
PEXPIRE key 60000          # 60000毫秒后过期

# 查看剩余时间
TTL key      # 秒级
PTTL key     # 毫秒级

# 取消过期（变为永久 Key）
PERSIST key

# 查看过期时间戳
EXPIRETIME key  # Redis 7.0+，返回 Unix 时间戳

jedis.set("token:user1", "abc123", "NX", "EX", 3600);
long ttl = jedis.ttl("token:user1"); // 秒
jedis.persist("token:user1"); // 取消过期

客户端写入请求
  → 检查当前内存使用量
  → 如果超过 maxmemory：
      a. 根据 maxmemory-policy 选择淘汰策略
      b. 选择要淘汰的 Key
      c. 删除 Key（同步或异步）
      d. 重试写入
  → 如果未超过 maxmemory：
      直接写入

Redis 维护一个主字典和一个较小的 LRU 字典：
  - 新写入的 Key 同时加入两个字典
  - 访问 Key 时从 LRU 字典中移动到"头部"
  - 需要淘汰时，从 LRU 字典随机采样 N 个 Key
  - 从采样结果中选择最久未使用的淘汰

淘汰时随机采样的 Key 数量（默认 5）
采样越多，越接近精确 LRU，但计算开销越大

# 设置内存上限
maxmemory 4gb

# 设置淘汰策略
maxmemory-policy allkeys-lru   # 推荐：全体 Key 按 LRU 淘汰

# 其他可选：
# maxmemory-policy allkeys-lfu    # Redis 4.0+，按访问频率
# maxmemory-policy volatile-lru   # 仅淘汰设置了过期的 Key
# maxmemory-policy noeviction     # 不淘汰，写入失败

# 调整采样数量
maxmemory-samples 5

# 查看当前配置
redis-cli CONFIG GET maxmemory-policy
redis-cli INFO memory

// 动态调整淘汰策略
jedis.configSet("maxmemory-policy", "allkeys-lru");
jedis.configSet("maxmemory", String.valueOf(4L * 1024 * 1024 * 1024));

// 监控内存使用
MemoryInfo memoryInfo = jedis.infoMemory();
long usedMemory = memoryInfo.getUsedMemory();

请求: GET /user?id=99999
  → Redis: miss
  → MySQL: SELECT * FROM user WHERE id=99999 → 空
  → 每次请求都打到 MySQL

热点 Key: product:1（库存 100）
  → 过期时间到，Key 失效
  → 1000 并发请求同时到达
  → 全部打到 MySQL，可能压垮数据库

场景一：大量 Key 同时过期
  → 系统在凌晨批量刷新缓存
  → 所有 Key 同时失效
  → 海量请求打到 MySQL

场景二：Redis 集群宕机
  → Redis 全部不可用
  → 所有请求打到 MySQL

方案一：布隆过滤器
  请求 → 布隆过滤器检查 → 可能存在 → Redis → DB
                         → 一定不存在 → 直接返回

方案二：空值缓存
  请求 → Redis miss → DB 查 → 空 → SET key "" EX 300
  后续请求命中空值缓存，直接返回空

方案三：参数校验
  对请求参数做合法性校验，非法参数直接拒绝

布隆过滤器是一个 bit 数组 + 多个哈希函数：
  1. 存入元素：用 K 个哈希函数计算 K 个位置，置 1
  2. 检查元素：用 K 个哈希函数计算 K 个位置
     → 所有位置都是 1 → 可能存在（有误判）
     → 任一位置是 0 → 一定不存在

方案一：互斥锁（分布式锁）
  请求 → Redis miss → 加锁（SETNX）→ 回源 DB → 写入 Redis → 释放锁
                     → 未获取锁 → 等待 → 重试 Redis

方案二：热点数据永不过期
  热点 Key 不设置过期时间
  异步线程定期更新

方案三：热点发现 + 主动预热
  监控访问频率，发现热点后自动续期

方案一：随机化过期时间
  基础过期时间 + 随机偏移
  → 避免大量 Key 同时失效

方案二：多级缓存
  → 本地缓存（Caffeine）+ Redis + DB
  → Redis 失效时本地缓存兜底

方案三：Redis 高可用
  → 主从复制 + 哨兵/Cluster
  → 实例故障自动切换

方案四：限流降级
  → 数据库层限流（每秒最大查询数）
  → 超过阈值返回默认值或降级

方案五：熔断降级
  → 检测到数据库压力过大
  → 自动熔断，返回缓存旧数据或默认值

// 使用 Redisson 的布隆过滤器
RBloomFilter<Long> bloomFilter = redisson.getBloomFilter("userIds");
bloomFilter.tryInit(1000000, 0.01); // 容量 100 万，误判率 1%
bloomFilter.add(1001L);
bloomFilter.contains(1001L); // true
bloomFilter.contains(9999L); // false（可能不存在）

public User getUser(Long id) {
    String key = "user:" + id;
    String cached = jedis.get(key);
    if (cached != null) {
        if ("".equals(cached)) {
            return null; // 空值缓存
        }
        return parseUser(cached);
    }
    User user = userMapper.selectById(id);
    if (user == null) {
        jedis.setex(key, 300, ""); // 缓存空值 5 分钟
    } else {
        jedis.setex(key, 3600, toJson(user));
    }
    return user;
}

public Product getProduct(Long id) {
    String key = "product:" + id;
    String lockKey = "lock:product:" + id;
    String cached = jedis.get(key);
    if (cached != null) {
        return parseProduct(cached);
    }

    // 加锁
    if (jedis.set(lockKey, "1", "NX", "EX", 10)) {
        try {
            // 二次检查
            cached = jedis.get(key);
            if (cached != null) {
                return parseProduct(cached);
            }
            // 回源数据库
            Product product = productMapper.selectById(id);
            if (product != null) {
                jedis.setex(key, 300, toJson(product));
            }
        } finally {
            jedis.del(lockKey); // 释放锁
        }
    } else {
        // 未获取锁，稍等重试
        Thread.sleep(50);
        return getProduct(id); // 递归重试
    }
    return null;
}

// 基础时间 + 随机偏移
int baseTtl = 3600; // 1 小时
int randomTtl = baseTtl + ThreadLocalRandom.current().nextInt(0, 1800);
jedis.setex(key, randomTtl, value);

系统启动 → 读取热点数据列表（从统计表或配置）
  → 批量查询数据库
  → 写入 Redis
  → 完成后开始接受请求

应用 → 订阅 MySQL Binlog → 解析变更事件
  → 更新 Redis 缓存
  → 实现缓存与数据库的准实时同步

定时任务（如每 5 分钟）
  → 扫描即将过期的热点 Key
  → 异步更新
  → 避免过期瞬间的缓存击穿

基于历史访问数据
  → 预测即将成为热点的数据
  → 提前加载到缓存
  → 适合有规律的热点场景（如早高峰新闻）

@Component
public class CacheWarmup implements ApplicationRunner {

    @Override
    public void run(ApplicationArguments args) {
        log.info("开始缓存预热...");

        // 1. 加载热点商品
        List<Long> hotProductIds = hotProductMapper.findHotIds(1000);
        for (Long id : hotProductIds) {
            Product product = productMapper.selectById(id);
            if (product != null) {
                jedis.setex("product:" + id, 3600, JSON.toJSONString(product));
            }
        }

        // 2. 加载热门配置
        List<Config> configs = configMapper.selectAll();
        for (Config config : configs) {
            jedis.setex("config:" + config.getKey(), 7200, config.getValue());
        }

        log.info("缓存预热完成，共加载 {} 个商品", hotProductIds.size());
    }
}

@Scheduled(fixedRate = 300000) // 每 5 分钟
public void refreshHotKeys() {
    Set<String> hotKeys = jedis.zrevrange("hot:key:ranking", 0, 999);
    for (String key : hotKeys) {
        if (jedis.ttl(key) < 300) { // 剩余时间 < 5 分钟
            refreshKeyFromDB(key);
        }
    }
}

#!/bin/bash
# 从数据库导出热点 ID
mysql -e "SELECT id FROM hot_products ORDER BY view_count DESC LIMIT 1000" > /tmp/hot_ids.txt

# 逐个加载到 Redis
while read id; do
    data=$(mysql -e "SELECT * FROM products WHERE id=$id" | tail -1)
    redis-cli SET "product:$id" "$data" EX 3600
done < /tmp/hot_ids.txt

echo "预热完成"

读流程：
  1. 读缓存
  2. 缓存命中 → 返回
  3. 缓存未命中 → 读数据库
  4. 写入缓存 → 返回

写流程：
  1. 写数据库
  2. 删除缓存（不是更新缓存）
  3. 完成

写流程：
  1. 写数据库
  2. 立即删除缓存
  3. 休眠（如 500ms）
  4. 再次删除缓存（防止并发回源导致的旧值写入）

写流程：
  1. 写数据库
  2. Binlog 产生变更记录
  3. Canal/Debezium 订阅 Binlog
  4. 解析变更，更新 Redis 缓存
  5. 实现准实时同步

@Service
public class UserService {

    @Autowired
    private Jedis jedis;

    public User getUser(Long id) {
        String key = "user:" + id;
        // 1. 读缓存
        String cached = jedis.get(key);
        if (cached != null) {
            return JSON.parseObject(cached, User.class);
        }
        // 2. 缓存未命中，读数据库
        User user = userMapper.selectById(id);
        if (user != null) {
            // 3. 写入缓存
            jedis.setex(key, 3600, JSON.toJSONString(user));
        }
        return user;
    }

    public void updateUser(User user) {
        // 1. 写数据库
        userMapper.updateById(user);
        // 2. 删除缓存（不是更新）
        jedis.del("user:" + user.getId());
    }
}

public void updateUserWithDoubleDelete(User user) {
    // 1. 写数据库
    userMapper.updateById(user);
    // 2. 立即删除缓存
    jedis.del("user:" + user.getId());
    // 3. 异步延时再删一次
    CompletableFuture.runAsync(() -> {
        try {
            Thread.sleep(500); // 延时
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        jedis.del("user:" + user.getId());
    });
}

DEL big_key（Hash，10000 元素）
  → 主线程执行
  → 遍历所有元素，逐一释放内存
  → 耗时可能达数百毫秒甚至数秒
  → 期间所有其他命令被阻塞

UNLINK big_key
  → 主线程：从数据库字典移除 Key（O(1)）
  → 后台线程：异步释放内存
  → 主线程不阻塞

redis-cli --bigkeys
  → 使用 SCAN 渐进遍历所有 Key
  → 检查每个 Key 的大小
  → 报告超过阈值的 Key

MEMORY USAGE key [SAMPLES count]
  → 估算指定 Key 的内存占用
  → SAMPLES：抽样数量（默认 5）
  → 对于大 Hash/ZSet，抽样估算更高效

# Redis 4.0+ 支持异步删除
redis-cli UNLINK big_key

# 查看异步删除进度
redis-cli INFO stats | grep async_deferred

# 扫描大 Key
redis-cli --bigkeys

# 限制扫描时间
redis-cli --bigkeys -i 0.1  # 每 100ms 暂停 10ms

// 方案一：Hash 拆分为多个小 Hash
// 原 Key: user:1:followers（10000 个粉丝）
// 拆分为：user:1:followers:0, user:1:followers:1, ..., user:1:followers:9
// 每个分片 1000 个

int batchSize = 1000;
for (int i = 0; i < totalFollowers; i += batchSize) {
    String shardKey = "user:1:followers:" + (i / batchSize);
    // HSET shardKey ...
}

// 方案二：List 拆分为多个 List
// 使用 LPUSH 分批写入

// 方案三：ZSet 按分数范围拆分
// rank:0-1000, rank:1001-2000, ...

@Scheduled(fixedRate = 600000) // 每 10 分钟
public void scanBigKeys() {
    String cursor = "0";
    do {
        ScanParams params = new ScanParams().count(500);
        ScanResult<String> result = jedis.scan(cursor, params);
        cursor = result.getCursor();
        for (String key : result.getResult()) {
            long size = jedis.memoryUsage(key);
            if (size > 10 * 1024) { // 超过 10KB
                log.warn("发现大 Key: {}, 大小: {}KB", key, size / 1024);
                // 通知或自动处理
            }
        }
    } while (!"0".equals(cursor));
}

碎片率 = used_memory_rss / used_memory

原因一：频繁创建和销毁 Key
  → 分配器分配内存时产生对齐间隙
  → 反复分配/释放后，内存出现碎片

原因二：jemalloc 分配器行为
  → jemalloc 按 size class 分配
  → 不同大小对象使用不同 size class
  → 对象释放后，相邻的空闲块可能无法合并

原因三：过期 Key 堆积
  → 大量过期 Key 未被及时清理
  → 占用的内存无法被复用

jemalloc 将内存划分为 arena：
  - 每个 arena 有多个 size class
  - 小对象（< 128KB）使用不同的 size class
  - 大对象使用独立的 page 分配
  - 碎片来自不同 size class 之间的内存无法复用

activedefrag（主动碎片整理）：
  → 后台线程定期扫描内存
  → 将分散的小对象移动到连续内存区域
  → 更新数据指针
  → 不阻塞主线程

redis-cli INFO memory
# 关注：
# used_memory: Redis 分配的内存
# used_memory_rss: 操作系统实际分配的内存
# mem_fragmentation_ratio: 碎片率

# 手动触发碎片整理（Redis 4.0+）
redis-cli MEMORY PURGE

# 查看碎片统计
redis-cli MEMORY DOCTOR

# 查看大 Key 分布
redis-cli MEMORY STATS

# Redis 配置
activedefrag yes
active-defrag-ignore-bytes 100mb     # 小于此值忽略
active-defrag-threshold-lower 10      # 碎片率低于此值不整理
active-defrag-threshold-upper 100     # 碎片率高于此值强制整理
active-defrag-cycle-min 5            # 整理周期最小值
active-defrag-cycle-max 25           # 整理周期最大值

redis-cli CONFIG GET activedefrag
redis-cli CONFIG SET activedefrag yes

客户端请求
  → L1 本地缓存（Caffeine）
     → 命中：返回（纳秒级）
     → 未命中：
  → L2 分布式缓存（Redis）
     → 命中：返回（毫秒级）+ 写入 L1
     → 未命中：
  → L3 数据库（MySQL）
     → 命中：返回 + 写入 L1 和 L2
     → 未命中：返回空

写流程：
  1. 写数据库
  2. 删除 L2（Redis）缓存
  3. 通知所有实例删除 L1（本地缓存）
     → 方案：Redis Pub/Sub、MQ 广播、版本号机制

方案一：Redis Pub/Sub
  写操作 → 发布 "cache:evict:{key}" → 所有实例订阅 → 删除本地缓存

方案二：版本号
  每个 Key 关联版本号
  写入时版本号 +1
  本地缓存存储版本号
  读取时检查版本号是否最新

方案三：短 TTL
  本地缓存设置较短 TTL（如 30s）
  自动过期保证最终一致

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import java.util.concurrent.TimeUnit;

// 创建本地缓存
Cache<String, User> localCache = Caffeine.newBuilder()
    .maximumSize(10_000)
    .expireAfterWrite(30, TimeUnit.SECONDS)
    .recordStats()
    .build();

// 读取
User user = localCache.get("user:" + id, key -> {
    // 缓存未命中，回源 Redis
    String cached = jedis.get(key);
    if (cached != null) {
        return JSON.parseObject(cached, User.class);
    }
    // Redis 也未命中，回源数据库
    User dbUser = userMapper.selectById(id);
    if (dbUser != null) {
        jedis.setex(key, 3600, JSON.toJSONString(dbUser));
    }
    return dbUser;
});

@Service
public class MultiLevelCache {

    private final Cache<String, Object> localCache; // L1
    private final Jedis jedis; // L2

    public <T> T get(String key, Class<T> type, Supplier<T> loader) {
        // L1
        Object cached = localCache.getIfPresent(key);
        if (cached != null) {
            return (T) cached;
        }
        // L2
        String redisVal = jedis.get(key);
        if (redisVal != null) {
            T value = JSON.parseObject(redisVal, type);
            localCache.put(key, value); // 回填 L1
            return value;
        }
        // L3
        T value = loader.get();
        if (value != null) {
            jedis.setex(key, 3600, JSON.toJSONString(value));
            localCache.put(key, value);
        }
        return value;
    }

    public void evict(String key) {
        localCache.invalidate(key);
        jedis.del(key);
    }
}

加锁：
  SET lock_key unique_value NX PX expire_time
  → NX：仅当 key 不存在时设置（保证互斥）
  → PX：设置过期时间（毫秒，防止死锁）

解锁：
  Lua 脚本保证原子性：
  if redis.call('GET', KEYS[1]) == ARGV[1] then
      return redis.call('DEL', KEYS[1])
  else
      return 0
  end
  → 先检查锁的持有者标识（防止误删别人的锁）
  → 再删除锁（原子操作）

public class SimpleRedisLock {

    private final Jedis jedis;

    public String lock(String key, long expireMs) {
        String requestId = UUID.randomUUID().toString();
        String result = jedis.set(key, requestId, "NX", "PX", expireMs);
        if ("OK".equals(result)) {
            return requestId;
        }
        return null; // 获取锁失败
    }

    public boolean unlock(String key, String requestId) {
        String luaScript =
            "if redis.call('GET', KEYS[1]) == ARGV[1] then " +
            "   return redis.call('DEL', KEYS[1]) " +
            "else return 0 end";
        Long result = jedis.eval(luaScript,
            Collections.singletonList(key),
            Collections.singletonList(requestId));
        return result == 1L;
    }
}

Config config = new Config();
config.useSingleServer()
    .setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);

RLock lock = redisson.getLock("myLock");

// 加锁（30s 自动过期，看门狗续期）
lock.lock();
try {
    // 临界区代码
    doSomething();
} finally {
    lock.unlock(); // 解锁
}

1. 生成唯一标识（UUID + 线程ID）
2. SET lock_key requestId NX PX expireTime
   → NX：key 不存在才设置（互斥保证）
   → PX：设置毫秒级过期（死锁避免）
3. 返回 OK → 加锁成功
4. 返回 nil → 加锁失败，重试或快速失败

if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
→ 先 GET 检查持有者标识
→ 匹配则 DEL（原子操作）
→ 不匹配则不删除（防止误删）

加锁时：
  HINCRBY lock_key:count threadId 1
  如果计数 == 1：
    SET lock_key threadId+count PX expireTime
  否则：
    续期

解锁时：
  HINCRBY lock_key:count threadId -1
  如果计数 == 0：
    DEL lock_key
    DEL lock_key:count

加锁成功后，启动看门狗线程：
  → 每隔 lockWatchdogTimeout/3（默认 10s）检查
  → 如果锁还被当前线程持有，续期到 lockWatchdogTimeout（默认 30s）
  → 如果锁已释放或 Redis 不可达，停止续期

public class RedisDistributedLock implements AutoCloseable {

    private static final String UNLOCK_SCRIPT =
        "if redis.call('GET', KEYS[1]) == ARGV[1] then " +
        "   return redis.call('DEL', KEYS[1]) " +
        "else return 0 end";

    private final Jedis jedis;
    private final String lockKey;
    private final String requestId;
    private final long expireMs;
    private final Timer watchdogTimer;

    public RedisDistributedLock(Jedis jedis, String lockKey, long expireMs) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.requestId = UUID.randomUUID() + ":" +
            Thread.currentThread().getId();
        this.expireMs = expireMs;
        this.watchdogTimer = new Timer(true);
    }

    public boolean tryLock(long timeoutMs) {
        long deadline = System.currentTimeMillis() + timeoutMs;
        while (System.currentTimeMillis() < deadline) {
            String result = jedis.set(lockKey, requestId,
                "NX", "PX", expireMs);
            if ("OK".equals(result)) {
                startWatchdog();
                return true;
            }
            try {
                Thread.sleep(50); // 退避
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }
        return false;
    }

    private void startWatchdog() {
        long period = expireMs / 3; // 每 1/3 过期时间续期
        watchdogTimer.schedule(new TimerTask() {
            @Override
            public void run() {
                String current = jedis.get(lockKey);
                if (requestId.equals(current)) {
                    jedis.pexpire(lockKey, expireMs);
                } else {
                    watchdogTimer.cancel();
                }
            }
        }, period, period);
    }

    public boolean unlock() {
        watchdogTimer.cancel();
        Long result = jedis.eval(UNLOCK_SCRIPT,
            Collections.singletonList(lockKey),
            Collections.singletonList(requestId));
        return result == 1L;
    }

    @Override
    public void close() {
        unlock();
    }
}

RedissonClient redisson = Redisson.create(config);

// 可重入锁
RLock lock = redisson.getLock("order:lock");
boolean acquired = lock.tryLock(3, 10, TimeUnit.SECONDS);
if (acquired) {
    try {
        // 业务逻辑
    } finally {
        lock.unlock();
    }
}

// 公平锁（按等待顺序获取）
RLock fairLock = redisson.getFairLock("fair:lock");

// 读写锁
RReadWriteLock rwLock = redisson.getReadWriteLock("rw:lock");
rwLock.writeLock().lock(); // 写锁
rwLock.readLock().lock();  // 读锁

SETNX key value
→ 仅当 key 不存在时，设置 key 的值为 value
→ 如果 key 已存在，不做任何操作
→ 返回 1：设置成功
→ 返回 0：key 已存在，未设置

SET key value [NX|XX] [EX seconds|PX milliseconds]
→ NX：仅 key 不存在时设置（等价于 SETNX + SET）
→ XX：仅 key 已存在时设置
→ EX/PX：设置过期时间
→ 原子操作，一步完成

步骤 1：SETNX lock_key value   → 加锁（成功返回 1）
步骤 2：EXPIRE lock_key 30     → 设置过期时间（30 秒）

问题场景：
  → 步骤 1 执行成功
  → 步骤 2 执行前，进程崩溃/OOM
  → lock_key 永不过期
  → 死锁！

SET lock_key value NX PX 30000
→ 一步完成：加锁 + 设置 30 秒过期
→ 原子操作，不存在中间状态
→ 即使进程崩溃，锁也会在 30 秒后自动释放

Redis 单线程执行：
  → SET 命令的所有参数在同一事件循环中执行
  → NX 检查和 PX 设置在同一个执行单元
  → 不会被其他命令中断

# 错误做法（不要用）
SETNX lock_key unique_value
EXPIRE lock_key 30

# 正确做法
SET lock_key unique_value NX PX 30000
# 或
SET lock_key unique_value NX EX 30

# 检查是否支持原子 SET
redis-cli INFO server | grep redis_version
# >= 2.6.12 支持

// 错误示范
if (jedis.setnx(lockKey, requestId) == 1) {
    jedis.expire(lockKey, 30); // 非原子！危险！
}

// 正确示范
String result = jedis.set(lockKey, requestId,
    "NX", "PX", 30000); // 原子操作
if ("OK".equals(result)) {
    // 加锁成功
}

加锁时设置过期时间：
  SET lock_key request_id NX PX 30000
  → 30 秒后 Redis 自动删除 lock_key
  → 即使进程崩溃，锁也会自动释放

Redis 内部实现：
  → lock_key 存入 expires 字典
  → 惰性删除 + 定期删除机制自动清理

看门狗线程：
  → 启动时记录锁的过期时间（如 30s）
  → 每隔 expire/3（如 10s）检查一次：
     a. 当前进程是否还持有锁？（GET 检查唯一标识）
     b. 如果是，续期到 30s（PEXPIRE）
     c. 如果否，停止续期
  → 业务完成后，主动停止看门狗

// Redisson 默认看门狗
private long lockWatchdogTimeout = 30 * 1000; // 30 秒

// 加锁成功后启动看门狗
protected void scheduleExpirationRenewal(long threadId) {
    RFuture<?> renewalFuture = new CompletableFuture();
    renewalFuture.addListener((r, i) -> {
        // 每 10 秒续期一次
        timeout(renewalFuture, lockWatchdogTimeout / 3,
            TimeUnit.MILLISECONDS);
    });
    // 续期逻辑
    renewalFuture = renewalFuture.thenCompose(
        res -> renewExpiration());
}

// 续期命令
RFuture<Boolean> renewExpiration() {
    return commandExecutor.evalWriteAsync(
        getLockKey(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
        // Lua 脚本：检查持有者 + 续期
        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
        "  redis.call('pexpire', KEYS[1], ARGV[1]); " +
        "  return 1; " +
        "end; " +
        "return 0;",
        Collections.<Object>singletonList(getLockKey()),
        lockWatchdogTimeout, getEntryName());
}

# 加锁时设置 30 秒过期
SET order:lock uuid-123 NX PX 30000

public class WatchdogLock {

    private final Jedis jedis;
    private final String lockKey;
    private final String ownerId;
    private final long expireMs;
    private ScheduledExecutorService scheduler;

    public boolean tryLock() {
        String result = jedis.set(lockKey, ownerId,
            "NX", "PX", expireMs);
        if ("OK".equals(result)) {
            startWatchdog();
            return true;
        }
        return false;
    }

    private void startWatchdog() {
        scheduler = Executors.newSingleThreadScheduledExecutor();
        long period = expireMs / 3;
        scheduler.scheduleAtFixedRate(() -> {
            String current = jedis.get(lockKey);
            if (ownerId.equals(current)) {
                jedis.pexpire(lockKey, expireMs);
            } else {
                scheduler.shutdown();
            }
        }, period, period, TimeUnit.MILLISECONDS);
    }

    public void unlock() {
        if (scheduler != null) {
            scheduler.shutdown();
        }
        // Lua 脚本原子解锁
    }
}

RLock lock = redisson.getLock("myLock");
// 默认开启看门狗，30 秒过期，每 10 秒续期
lock.lock(); // 无参 lock() 开启看门狗

// 如果指定超时时间，看门狗不启用
lock.lock(10, TimeUnit.SECONDS); // 看门狗不启用

方法 A() {
    lock.lock(); // 第 1 次加锁
    B();
    lock.unlock(); // 第 1 次解锁
}

方法 B() {
    lock.lock(); // 第 2 次加锁（同一线程，可重入）
    // ...
    lock.unlock(); // 第 2 次解锁
}

Redis 数据结构：
  lock_key → threadId:count（Hash 结构）

加锁流程（Lua 脚本）：
  1. 检查 lock_key 是否存在
  2. 如果不存在：
     a. SET lock_key threadId+identifier PX expireTime
     b. HSET lock_key threadId count=1
     c. 返回加锁成功
  3. 如果存在，检查当前线程是否已持有：
     a. HEXISTS lock_key threadId
     b. 如果存在（可重入）：
        HINCRBY lock_key threadId +1
        续期
        返回加锁成功
     c. 如果不存在（其他线程持有）：
        返回加锁失败，等待

解锁流程（Lua 脚本）：
  1. 检查当前线程是否持有锁
  2. HINCRBY lock_key threadId -1
  3. 如果计数变为 0：
     DEL lock_key（真正释放锁）
     DEL lock_key:count
  4. 如果计数 > 0：
     仅更新计数（未完全释放）

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])

if (redis.call('hexists', KEYS[1], ARGV[2]) == 0) then
    return nil -- 非持有者，不能解锁
end
local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1)
if (counter > 0) then
    redis.call('pexpire', KEYS[1], ARGV[1])
    return 0 -- 未完全释放
else
    redis.call('del', KEYS[1])
    return 1 -- 完全释放
end

RLock lock = redisson.getLock("myLock");

// 第 1 次加锁
lock.lock();
try {
    // 第 2 次加锁（可重入）
    lock.lock();
    try {
        // 业务逻辑
    } finally {
        lock.unlock(); // 第 2 次解锁
    }
} finally {
    lock.unlock(); // 第 1 次解锁
}

public class ReentrantRedisLock {

    private final Jedis jedis;
    private final String lockKey;
    private final long expireMs;
    private final ThreadLocal<String> threadIdHolder =
        ThreadLocal.withInitial(() ->
            Thread.currentThread().getId() + ":" +
            UUID.randomUUID().toString());

    private static final String LOCK_SCRIPT =
        "if (redis.call('exists', KEYS[1]) == 0) then " +
        "  redis.call('hset', KEYS[1], ARGV[2], 1) " +
        "  redis.call('pexpire', KEYS[1], ARGV[1]) " +
        "  return 0 " +
        "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 0 " +
        "end " +
        "return redis.call('pttl', KEYS[1])";

    private static final String UNLOCK_SCRIPT =
        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 0) then " +
        "  return nil " +
        "end " +
        "local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1) " +
        "if (counter > 0) then " +
        "  redis.call('pexpire', KEYS[1], ARGV[1]) " +
        "  return 0 " +
        "else " +
        "  redis.call('del', KEYS[1]) " +
        "  return 1 " +
        "end";

    public boolean tryLock() {
        String threadId = threadIdHolder.get();
        Long result = jedis.eval(LOCK_SCRIPT,
            Collections.singletonList(lockKey),
            String.valueOf(expireMs), threadId);
        return result != null && result == 0L;
    }

    public void unlock() {
        String threadId = threadIdHolder.get();
        jedis.eval(UNLOCK_SCRIPT,
            Collections.singletonList(lockKey),
            String.valueOf(expireMs), threadId);
    }
}

lock() 调用：
  1. 检查当前线程是否已持有锁（可重入检查）
  2. 如果已持有 → 计数 +1，续期，返回
  3. 如果未持有 → Lua 脚本原子加锁：
     a. 锁不存在 → 创建锁，计数设为 1，设置过期
     b. 锁已存在且属于当前线程 → 计数 +1
     c. 锁已存在且属于其他线程 → 返回剩余过期时间
  4. 加锁成功 → 启动看门狗续期
  5. 加锁失败 → 订阅解锁消息 → 阻塞等待 → 收到通知后重试

1. 加锁成功后，启动续期任务
2. 每 lockWatchdogTimeout/3（默认 10s）执行一次：
   a. Lua 脚本检查锁是否被当前线程持有
   b. 如果是 → PEXPIRE 续期到 30s
   c. 如果否 → 停止续期
3. 解锁时 → 停止续期任务

1. 依次向 N 个独立 Redis 实例加锁（默认 5 个）
2. 如果加锁成功数 > N/2 + 1（多数派）：
   → 认为加锁成功
   → 持有锁直到业务完成
3. 如果加锁成功数 <= N/2：
   → 向所有已加锁的实例解锁
   → 认为加锁失败

// 配置
Config config = new Config();
config.useClusterServers()
    .addNodeAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);

// 可重入锁
RLock lock = redisson.getLock("order:lock");
lock.lock(10, 30, TimeUnit.SECONDS);
// 参数：10s 等待时间，30s 持有时间，时间单位
try {
    // 业务逻辑
} finally {
    lock.unlock();
}

// 公平锁
RLock fairLock = redisson.getFairLock("fair:lock");
fairLock.lock();

// 读写锁
RReadWriteLock rwLock = redisson.getReadWriteLock("rw:lock");
rwLock.readLock().lock();  // 读锁（可共享）
rwLock.writeLock().lock(); // 写锁（互斥）

// 信号量
RSemaphore semaphore = redisson.getSemaphore("permits");
semaphore.acquire(3); // 获取 3 个许可
semaphore.release(2); // 释放 2 个许可

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
            .setAddress("redis://127.0.0.1:6379")
            .setPassword("password");
        return Redisson.create(config);
    }
}

@Service
public class OrderService {

    @Autowired
    private RedissonClient redisson;

    public void createOrder(Long productId) {
        RLock lock = redisson.getLock("stock:" + productId);
        try {
            if (lock.tryLock(3, 10, TimeUnit.SECONDS)) {
                // 扣减库存
            }
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

1. 客户端 A 在主节点加锁成功
2. 主节点将锁同步到从节点之前崩溃
3. 从节点升级为主节点
4. 客户端 B 在新主节点加锁成功（因为旧锁没同步过来）
5. 两个客户端同时持有锁 → 锁失效

1. 获取当前时间 T1
2. 依次向 N 个独立 Redis 实例加锁：
   SET lock_key unique_value NX PX expire
   → 每次加锁设置较短的超时（如 50ms）
3. 统计加锁成功的实例数
4. 计算总耗时：T2 - T1
5. 判断：
   a. 成功数 > N/2 + 1（多数派）
   b. 总耗时 < 锁过期时间
   → 两个条件都满足 → 加锁成功
   → 否则 → 向所有已加锁的实例解锁，加锁失败

1. 依次向所有 N 个 Redis 实例发送解锁命令
2. 不关心个别实例的解锁结果
3. 只要加锁时多数成功，解锁时尽力而为

加锁失败后：
  → 随机延迟一段时间（如 10-50ms）
  → 重新尝试
  → 避免所有客户端同时重试导致"惊群效应"

实例数量 N：建议 5 个（奇数，便于多数派判断）
加锁超时：每个实例 50ms（网络延迟上限）
锁过期时间：30s（业务最大执行时间 + 缓冲）
重试间隔：随机 10-50ms

// 创建多个 RedissonClient（每个对应独立的 Redis 实例）
RedissonClient redisson1 = Redisson.create(config1);
RedissonClient redisson2 = Redisson.create(config2);
RedissonClient redisson3 = Redisson.create(config3);
RedissonClient redisson4 = Redisson.create(config4);
RedissonClient redisson5 = Redisson.create(config5);

// 创建 Redlock
RedissonRedLock redLock = new RedissonRedLock(
    redisson1.getLock("lock_key"),
    redisson2.getLock("lock_key"),
    redisson3.getLock("lock_key"),
    redisson4.getLock("lock_key"),
    redisson5.getLock("lock_key")
);

// 加锁
boolean acquired = redLock.tryLock(30, 10, TimeUnit.SECONDS);
if (acquired) {
    try {
        // 业务逻辑
    } finally {
        redLock.unlock();
    }
}

public class Redlock {

    private final List<Jedis> jedisList; // 多个独立 Redis 实例
    private static final int QUORUM = 3; // 多数派（5 个实例中需 3 个成功）
    private static final long LOCK_TIMEOUT = 30000; // 30 秒
    private static final int INSTANCE_TIMEOUT = 50; // 单实例超时 50ms

    public boolean tryLock(String lockKey, String value) {
        long startTime = System.currentTimeMillis();
        int successCount = 0;

        for (Jedis jedis : jedisList) {
            try {
                String result = jedis.set(lockKey, value,
                    "NX", "PX", LOCK_TIMEOUT);
                if ("OK".equals(result)) {
                    successCount++;
                }
            } catch (Exception e) {
                // 单个实例异常，忽略
            }
        }

        long elapsed = System.currentTimeMillis() - startTime;
        boolean success = successCount >= QUORUM
            && elapsed < LOCK_TIMEOUT;

        if (!success) {
            unlockAll(lockKey, value); // 失败则清理
        }
        return success;
    }

    public void unlockAll(String lockKey, String value) {
        String luaScript =
            "if redis.call('GET', KEYS[1]) == ARGV[1] then " +
            "  return redis.call('DEL', KEYS[1]) " +
            "else return 0 end";
        for (Jedis jedis : jedisList) {
            try {
                jedis.eval(luaScript,
                    Collections.singletonList(lockKey),
                    Collections.singletonList(value));
            } catch (Exception e) {
                // 忽略单个实例的解锁异常
            }
        }
    }
}

Client → Master（主节点，读写）
           ↓ 异步复制
         Slave1（从节点，只读）
         Slave2（从节点，只读）

1. 从节点执行 SLAVEOF host port
2. 从节点向主节点发送 PSYNC 命令（携带 runId 和 offset）
3. 主节点判断是否为首次同步：
   a. 是首次 → 执行 BGSAVE 生成 RDB
   b. 不是首次 → 尝试部分同步（offset 之后的命令）
4. 主节点 BGSAVE 完成后：
   a. 将 RDB 文件发送给从节点
   b. 从节点接收并加载 RDB
5. 主节点将 BGSAVE 期间产生的增量命令发送给从节点
6. 从节点执行增量命令，同步完成

1. 主节点执行写命令（SET、HSET 等）
2. 命令写入复制缓冲区（replication backlog）
3. 主节点将缓冲区内容异步发送给从节点
4. 从节点接收并执行命令
5. 从节点记录同步偏移量（offset）
6. 断线重连时，从节点发送 PSYNC 携带 offset
   → 主节点从缓冲区中读取 offset 之后的命令
   → 实现部分同步（无需全量复制）

主节点维护一个环形缓冲区：
  → 存储最近的写命令
  → 默认大小 1MB（可配置 repl-backlog-size）
  → 记录每个命令的 offset
  → 从节点断线重连时，主节点从缓冲区读取增量命令

如果缓冲区被覆盖（从节点断线时间过长）：
  → 无法部分同步
  → 必须重新全量复制

每个主节点有两个复制 ID：
  - replicationid：主节点自身的 ID（40 位随机字符串）
  - replicationid2：上一任主节点的 ID（用于故障恢复）

从节点保存主节点的 replicationid：
  → PSYNC 时携带这个 ID
  → 主节点判断是否是同一个主节点

主节点：
  → 每 10 秒向所有从节点发送 PING
  → 检查从节点是否存活
  → 记录每个从节点的延迟时间

从节点：
  → 每 1 秒向主节点发送 REPLCONF ACK
  → 报告同步进度（offset）
  → 主节点据此判断从节点延迟

Redis 使用异步复制，存在短暂不一致：
  → 写命令在主节点执行后立即返回客户端
  → 不等待从节点确认
  → 主节点宕机时，最近的写操作可能未同步到从节点

缓解措施：
  → 配置 min-slaves-to-write：至少 N 个从节点连接时才接受写
  → 配置 min-slaves-max-lag：从节点最大延迟时间
  → 这两个配置实现"准同步"复制

# 从节点配置（redis.conf）
replicaof 127.0.0.1 6379   # Redis 5.0 之前用 slaveof
# 或运行时配置
redis-cli SLAVEOF 127.0.0.1 6379

# 主节点查看复制状态
redis-cli INFO replication
# 查看：connected_slaves、master_link_status、offset 等

# 从节点取消复制
redis-cli SLAVEOF NO ONE

# 设置复制缓冲区大小
repl-backlog-size 64mb      # 默认 1MB，可增大到 64MB

# 设置无从节点时是否保留缓冲区
repl-backlog-ttl 3600       # 无从节点 1 小时后释放缓冲区

# 至少 2 个从节点连接时才接受写
min-replicas-to-write 2
# 从节点最大延迟 10 秒
min-replicas-max-lag 10

# 如果不满足条件，主节点拒绝写操作
# 保护数据不丢失，但可能导致写入失败

@Service
public class RedisService {

    @Autowired
    private Jedis masterJedis;   # 主节点

    @Autowired
    private Jedis slaveJedis;    # 从节点

    // 写操作走主节点
    public void set(String key, String value) {
        masterJedis.set(key, value);
    }

    // 读操作走从节点
    public String get(String key) {
        return slaveJedis.get(key);
    }
}

                    ┌──────────────┐
                    │  Sentinel 1  │
                    └──────┬───────┘
                           │ 监控
                    ┌──────┴───────┐
                    │  Sentinel 2  │
                    └──────┬───────┘
                           │
                    ┌──────┴───────┐
                    │  Sentinel 3  │
                    └──────┬───────┘
                           │ 监控
              ┌────────────┼────────────┐
              │            │            │
        ┌─────┴─────┐┌────┴────┐┌─────┴────┐
        │  Master   ││ Slave1  ││ Slave2  │
        └───────────┘└─────────┘└──────────┘

1. 哨兵每 2 秒向所有 Redis 节点发送 PING
2. 如果节点在 timeout（默认 30s）内无响应：
   → 标记为"主观下线"（SDOWN）
3. 如果多数哨兵（N/2+1）都认为节点下线：
   → 标记为"客观下线"（ODOWN）
   → 触发故障转移

1. 当主节点被标记为客观下线时
2. 所有哨兵节点进行投票选举：
   a. 每个哨兵向其他哨兵发送投票请求
   b. 获得多数票（N/2+1）的哨兵成为 Leader
   c. Leader 负责执行故障转移
3. 如果选举失败（平票）：
   → 等待一个随机时间后重新选举

1. Leader 选举新主节点：
   a. 从从节点中选择数据最新的（复制 offset 最大）
   b. 向新主节点发送 SLAVEOF NO ONE（升级为主节点）
2. Leader 重新配置其他从节点：
   a. 向其他从节点发送 SLAVEOF new_host new_port
   b. 从节点切换到新主节点
3. 通知所有客户端新主节点地址

sentinel monitor mymaster 127.0.0.1 6379 2
# 参数：监控名称 主节点地址 端口 法定人数（几个哨兵同意判定下线）

sentinel down-after-milliseconds mymaster 30000
# 主观下线超时时间（毫秒）

sentinel failover-timeout mymaster 60000
# 故障转移超时时间

sentinel parallel-syncs mymaster 1
# 故障转移后，同时向新主节点同步的从节点数量

哨兵使用简化版 Raft 协议：
  → 每个哨兵有一个 term（任期编号）
  → 选举时递增 term
  → 获得多数票者当选 Leader
  → Leader 在任期内负责故障转移

关键参数：
  → 选举超时：15s（随机）
  → 心跳间隔：2s
  → 法定人数：N/2+1（N 为哨兵总数）

# 1. 启动主节点
redis-server redis-master.conf

# 2. 启动两个从节点
redis-server redis-slave1.conf  # replicaof master 6379
redis-server redis-slave2.conf  # replicaof master 6379

# 3. 启动三个哨兵
redis-server sentinel-1.conf
redis-server sentinel-2.conf
redis-server sentinel-3.conf

# sentinel.conf 配置：
# sentinel monitor mymaster 127.0.0.1 6379 2
# sentinel down-after-milliseconds mymaster 30000
# sentinel failover-timeout mymaster 60000

Set<String> sentinels = new HashSet<>(Arrays.asList(
    "127.0.0.1:26379",
    "127.0.0.1:26380",
    "127.0.0.1:26381"
));

JedisPoolConfig poolConfig = new JedisPoolConfig();
JedisSentinelPool pool = new JedisSentinelPool(
    "mymaster", sentinels, poolConfig);

Jedis jedis = pool.getResource();
jedis.set("key", "value");
jedis.close();

Client → (CRC16 路由)
  → Master1 (槽 0-5460) → Slave1
  → Master2 (槽 5461-10922) → Slave2
  → Master3 (槽 10923-16383) → Slave3

所有 Key 被映射到 16384 个槽：
  slot = CRC16(key) % 16384

示例：
  Key "user:1" → CRC16 → slot 1234 → Master1
  Key "product:1" → CRC16 → slot 9999 → Master2

每个主节点负责一部分槽：
  Master1: 槽 0-5460（约 1/3 数据）
  Master2: 槽 5461-10922（约 1/3 数据）
  Master3: 槽 10923-16383（约 1/3 数据）

CRC16（Cyclic Redundancy Check 16-bit）：
  → 16 位循环冗余校验算法
  → 将任意长度的 Key 映射到 0-65535
  → CRC16(key) % 16384 → 0-16383（16384 个槽）

Key 的哈希 tag：
  → 用 { } 包裹 Key 的一部分，只对这部分计算 CRC16
  → 保证相关的 Key 落在同一槽

示例：
  user:{1}:name → CRC16("1") → 所有 {1} 的 Key 在同一槽
  user:{1}:email → 同上
  user:{2}:name → CRC16("2") → 在另一个槽

节点间通过 Gossip 协议通信：
  → 每个节点周期性地与其他节点交换状态信息
  → 信息包括：节点状态、槽分配、主从关系
  → 最终一致性，无需中心节点

集群元数据：
  → 每个节点都保存完整的集群状态
  → 客户端连接任何节点都可获得路由信息

1. 客户端连接任意集群节点
2. 发送命令（如 SET key value）
3. 节点计算 key 所属槽：
   slot = CRC16(key) % 16384
4. 如果该槽由本节点负责：
   → 直接执行命令
   → 返回结果
5. 如果该槽由其他节点负责：
   → 返回 MOVED 重定向
   → 客户端连接正确的节点重试
6. 客户端缓存路由表，后续请求直接路由

1. 新增节点加入集群
2. 从现有主节点迁移槽到新节点：
   a. 源主节点：接收迁移命令
   b. 迁移数据到目标主节点
   c. 更新集群配置
   d. 通知所有节点
3. 槽迁移完成，新节点开始服务

1. 从节点检测到主节点故障
2. 从节点发起故障转移选举
3. 集群中多数主节点投票
4. 获得投票的从节点升级为新主节点
5. 所有节点更新槽分配
6. 客户端通过 MOVED 重定向发现新主节点

# 1. 启动 6 个 Redis 实例（3 主 3 从）
redis-server node1.conf  # port 6379, cluster-enabled yes
redis-server node2.conf  # port 6380
...
redis-server node6.conf  # port 6384

# 2. 创建集群
redis-cli --cluster create \
  127.0.0.1:6379 127.0.0.1:6380 127.0.0.1:6381 \
  127.0.0.1:6382 127.0.0.1:6383 127.0.0.1:6384 \
  --cluster-replicas 1
# 参数：前三个为主节点，后三个为从节点

Set<HostAndPort> clusterNodes = new HashSet<>();
clusterNodes.add(new HostAndPort("127.0.0.1", 6379));
clusterNodes.add(new HostAndPort("127.0.0.1", 6380));
clusterNodes.add(new HostAndPort("127.0.0.1", 6381));

JedisCluster jedisCluster = new JedisCluster(clusterNodes);

// 写操作（自动路由）
jedisCluster.set("user:1", "Alice");
// 读操作
String name = jedisCluster.get("user:1");
// Hash 操作
jedisCluster.hset("user:1", "age", "25");

# 查看集群信息
redis-cli CLUSTER INFO
redis-cli CLUSTER NODES

# 查看槽分配
redis-cli CLUSTER SLOTS

# 手动迁移槽
redis-cli CLUSTER SETSLOT <slot> <node_id>
redis-cli CLUSTER MIGRATE <host> <port> <key> <dest_db> <timeout>

# 添加节点
redis-cli CLUSTER MEET <host> <port>

# 移除节点
redis-cli CLUSTER FORGET <node_id>
redis-cli CLUSTER REPLICATE <master_node_id>

// 保证相关 Key 在同一槽
// 使用 { } 包裹固定部分
jedisCluster.set("{user:1}:name", "Alice");
jedisCluster.set("{user:1}:email", "alice@example.com");
// 这两个 Key 在同一槽，可以使用事务或 Lua 脚本

生产者 → LPUSH queue task_json
消费者 → BRPOP queue 0（阻塞等待）
       → 处理任务
       → 任务自动移除

可靠队列（Stream）：
生产者 → XADD stream task
消费者 → XREADGROUP group stream
       → 处理后 ACK
       → 支持消费组、消息持久化、回溯

计数器：
  INCR counter  → 原子递增
  EXPIRE counter 60 → 设置过期时间（1 分钟窗口）

限流（固定窗口）：
  SET key count EX 60
  INCR key → 如果超过阈值则拒绝

限流（滑动窗口 - Lua 实现）：
  → ZADD 记录每个请求的时间戳
  → ZREMRANGEBYSCORE 清除过期记录
  → ZCARD 统计当前窗口内请求数
  → 超过阈值则拒绝

限流（令牌桶）：
  → 按固定速率往桶中添加令牌
  → 请求消费令牌，无令牌则拒绝

发布者 → PUBLISH channel message
订阅者 → SUBSCRIBE channel
       → 收到消息推送
       → 实时性高但不持久化

// 生产者
jedis.lpush("order:queue", orderJson);

// 消费者
List<String> result = jedis.brpop(0, "order:queue");
String orderJson = result.get(1);
// 处理订单
processOrder(orderJson);

// 生产者
jedis.xadd("stream:orders",
    StreamAddParams.map("type", "CREATE", "data", orderJson));

// 消费者组
jedis.xgroupCreate("stream:orders", "orderGroup",
    StreamGroupParams.xgroupParams().sentinel(true));

// 消费消息
List<StreamEntry> messages = jedis.xreadGroup(
    "orderGroup", "consumer1",
    new HashMap<String, StreamEntryID>() {{
        put("stream:orders", StreamEntryID.last());
    }}, 10);

// 确认消费
jedis.xack("stream:orders", "orderGroup",
    message.getId());

-- KEYS[1]: 限流 Key
-- ARGV[1]: 当前时间戳
-- ARGV[2]: 窗口大小（毫秒）
-- ARGV[3]: 最大请求数

local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

-- 清除窗口外的记录
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)

-- 统计当前窗口内请求数
local count = redis.call('ZCARD', key)

if count < limit then
    -- 添加新请求
    redis.call('ZADD', key, now, now .. '-' .. math.random())
    redis.call('PEXPIRE', key, window)
    return 1  -- 允许
else
    return 0  -- 限流
end

// 生产者：添加延迟任务，分数为执行时间戳
long executeTime = System.currentTimeMillis() + 5000; // 5 秒后
jedis.zadd("delay:queue", executeTime, "task:1");

// 消费者：定时扫描到期任务
@Scheduled(fixedRate = 1000) // 每秒扫描
public void processDelayQueue() {
    long now = System.currentTimeMillis();
    Set<String> tasks = jedis.zrangeByScore("delay:queue", 0, now);
    for (String task : tasks) {
        if (jedis.zrem("delay:queue", task) > 0) { // 原子移除
            // 执行任务
            processTask(task);
        }
    }
}

1. 请求到达，计算 key（接口+IP+时间窗口）
2. Lua 脚本：
   a. ZREMRANGEBYSCORE 清除窗口外记录
   b. ZCARD 统计窗口内请求数
   c. 如果 < 阈值：ZADD 记录请求，返回允许
   d. 如果 >= 阈值：返回拒绝
3. Lua 脚本保证原子性

生产者：
  ZADD delay:queue executeTime taskId

消费者（定时扫描）：
  1. 每秒执行 SCAN
  2. ZRANGEBYSCORE 获取到期任务
  3. ZREM 原子移除
  4. 执行任务

生产者：
  XADD stream field value
  → 消息持久化

消费者组：
  XREADGROUP group stream
  → 读取未消费的消息
  → 未确认的消息可重复消费

确认：
  XACK stream group messageId
  → 标记消息已消费

-- rate_limit.lua
-- KEYS[1]: 限流 key
-- ARGV[1]: 当前时间戳（毫秒）
-- ARGV[2]: 窗口大小（毫秒）
-- ARGV[3]: 最大请求数

local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    redis.call('ZADD', key, now, now .. ':' .. math.random(1000000))
    redis.call('PEXPIRE', key, window)
    return 1
else
    return 0
end

public class RateLimiter {

    private static final String LUA_SCRIPT = "...";

    public boolean isAllowed(String api, String ip, int windowMs, int maxRequests) {
        String key = "rate:" + api + ":" + ip;
        long now = System.currentTimeMillis();
        Long result = jedis.eval(LUA_SCRIPT,
            Collections.singletonList(key),
            String.valueOf(now),
            String.valueOf(windowMs),
            String.valueOf(maxRequests));
        return result == 1L;
    }
}

// 使用
if (rateLimiter.isAllowed("/api/order", "192.168.1.1", 1000, 10)) {
    // 允许请求
} else {
    // 返回 429 Too Many Requests
}

@Component
public class DelayQueue {

    @Autowired
    private Jedis jedis;

    public void addTask(String task, long delayMs) {
        long executeTime = System.currentTimeMillis() + delayMs;
        jedis.zadd("delay:queue", executeTime, task);
    }

    @Scheduled(fixedRate = 1000) // 每秒扫描
    public void processTasks() {
        long now = System.currentTimeMillis();
        Set<String> tasks = jedis.zrangeByScore("delay:queue", 0, now);
        for (String task : tasks) {
            // 原子移除并处理
            if (jedis.zrem("delay:queue", task) > 0) {
                processTask(task);
            }
        }
    }

    private void processTask(String task) {
        // 处理延迟任务
    }
}

public class ReliableMQ {

    @Autowired
    private Jedis jedis;

    public void send(String topic, Map<String, String> message) {
        jedis.xadd("stream:" + topic,
            StreamAddParams.map(message));
    }

    public void createConsumerGroup(String topic, String group) {
        try {
            jedis.xgroupCreate("stream:" + topic, group,
                StreamGroupParams.xgroupParams());
        } catch (JedisBusyException e) {
            // 消费组已存在
        }
    }

    public List<StreamEntry> consume(String topic, String group,
                                       String consumer, int count) {
        return jedis.xreadGroup(
            group, consumer,
            new HashMap<String, StreamEntryID>() {{
                put("stream:" + topic, StreamEntryID.last());
            }},
            count);
    }

    public void ack(String topic, String group, StreamEntryID id) {
        jedis.xack("stream:" + topic, group, id);
    }
}

库存 = 1
请求 A: 读库存 = 1 → 准备扣减
请求 B: 读库存 = 1 → 准备扣减
请求 A: 扣减 → 库存 = 0
请求 B: 扣减 → 库存 = -1（超卖！）

// 错误做法（非原子）
int stock = Integer.parseInt(jedis.get("product:stock:1001"));
if (stock > 0) {
    jedis.decr("product:stock:1001");
    // 可能已被其他线程扣减
}

-- 扣减库存
-- KEYS[1]: 库存 key
-- ARGV[1]: 扣减数量
-- 返回：扣减后的库存，如果库存不足返回 -1

local key = KEYS[1]
local decrement = tonumber(ARGV[1])
local stock = tonumber(redis.call('GET', key))

if stock == nil then
    return -1  -- 库存不存在
end

if stock < decrement then
    return -1  -- 库存不足
end

redis.call('DECRBY', key, decrement)
return stock - decrement

-- KEYS[1]: 库存 key
-- KEYS[2]: 订单记录 key
-- ARGV[1]: 扣减数量
-- ARGV[2]: 订单 ID
-- ARGV[3]: 用户 ID

local stockKey = KEYS[1]
local orderKey = KEYS[2]
local decrement = tonumber(ARGV[1])
local orderId = ARGV[2]
local userId = ARGV[3]

-- 检查库存
local stock = tonumber(redis.call('GET', stockKey))
if stock == nil or stock < decrement then
    return -1  -- 库存不足
end

-- 扣减库存
redis.call('DECRBY', stockKey, decrement)

-- 记录订单（Hash 结构）
redis.call('HSET', orderKey, orderId, userId)

-- 设置过期时间（30 分钟）
redis.call('EXPIRE', orderKey, 1800)

return 1  -- 扣减成功

1. Redis 中设置预扣库存
2. 秒杀请求直接扣减 Redis 预扣库存
3. 异步将扣减结果同步到数据库
4. 用户支付成功 → 确认扣减
5. 用户超时未支付 → 回滚预扣

-- KEYS[1]: 库存 key
-- KEYS[2]: 订单记录 key
-- KEYS[3]: 去重 key
-- ARGV[1]: 扣减数量
-- ARGV[2]: 订单 ID

-- 检查是否已处理（防止重复提交）
if redis.call('SETNX', KEYS[3], ARGV[2]) == 0 then
    return -2  -- 重复请求
end

-- 扣减库存
-- ...（同上）

public class StockService {

    private static final String STOCK_LUA =
        "local stock = tonumber(redis.call('GET', KEYS[1])) " +
        "if stock == nil or stock < tonumber(ARGV[1]) then " +
        "  return -1 " +
        "end " +
        "redis.call('DECRBY', KEYS[1], ARGV[1]) " +
        "redis.call('SADD', KEYS[2], ARGV[2]) " +
        "return 1";

    @Autowired
    private Jedis jedis;

    public boolean deductStock(Long productId, int count,
                                 String orderId, String userId) {
        String stockKey = "product:stock:" + productId;
        String orderKey = "orders:" + productId;
        String dedupKey = "dedup:" + orderId;

        Long result = jedis.eval(STOCK_LUA,
            Arrays.asList(stockKey, orderKey, dedupKey),
            String.valueOf(count), orderId, userId);

        return result == 1L;
    }
}

RAtomicLong stock = redisson.getAtomicLong("product:stock:1001");
long remaining = stock.addAndGet(-1); // 原子扣减
if (remaining < 0) {
    stock.incrementAndGet(); // 回滚
    throw new StockNotEnoughException();
}

方案一：异步校验
  1. Redis 扣减成功
  2. 异步写入数据库订单
  3. 定时任务对账：Redis 订单 vs 数据库订单
  4. 不一致时补偿

方案二：可靠消息
  1. Redis 扣减
  2. 发送消息到 MQ
  3. 消费者处理消息，扣减数据库
  4. 消费失败重试

写入方：
  1. 写数据库
  2. 更新自己的本地缓存
  3. 发布缓存失效消息到 Redis Channel
     → PUBLISH cache:evict:{key} {key}

所有实例：
  1. 订阅失效 Channel
     → SUBSCRIBE cache:evict:* (使用模式订阅)
  2. 收到消息后删除本地缓存
     → localCache.invalidate(key)

优点：实时性高，变更立即通知
缺点：Pub/Sub 不持久化，新订阅的实例收不到历史消息

写入方：
  1. 写数据库
  2. INCR cache:version:{key}（版本号 +1）
  3. 更新自己的本地缓存（带版本号）

读取方：
  1. 从本地缓存读取数据 + 版本号
  2. 从 Redis 读取最新版本号
  3. 如果版本号一致 → 返回缓存数据
  4. 如果版本号不一致 → 回源数据库，更新缓存

优点：可靠，不丢消息
缺点：每次读多一次 Redis 请求

本地缓存设置较短 TTL（30 秒）：
  → 最长 30 秒后所有实例会回源刷新
  → 最终一致性

优点：实现简单
缺点：存在最长 TTL 时间的不一致窗口

1. 主方案：Pub/Sub 实时通知
   → 写入方发布失效消息
   → 所有实例实时删除本地缓存

2. 兜底方案：短 TTL（30 秒）
   → Pub/Sub 遗漏的消息（如实例重启期间）
   → 最长 30 秒后自动刷新

@Component
public class LocalCacheManager {

    private final Cache<String, Object> localCache = Caffeine.newBuilder()
        .maximumSize(10_000)
        .expireAfterWrite(30, TimeUnit.SECONDS) // 兜底 TTL
        .build();

    private final Jedis jedis;

    @PostConstruct
    public void init() {
        // 订阅缓存失效消息
        new Thread(() -> {
            jedis.psubscribe(new JedisPubSub() {
                @Override
                public void onPMessage(String pattern, String channel,
                                       String message) {
                    localCache.invalidate(message);
                }
            }, "cache:evict:*");
        }).start();
    }

    public void evictCache(String key) {
        // 删除自己的本地缓存
        localCache.invalidate(key);
        // 通知其他实例
        jedis.publish("cache:evict:" + key, key);
    }

    public Object get(String key) {
        Object value = localCache.getIfPresent(key);
        if (value != null) {
            return value;
        }
        // 回源数据库
        value = loadFromDB(key);
        if (value != null) {
            localCache.put(key, value);
        }
        return value;
    }

    public void update(String key, Object value) {
        // 1. 写数据库
        saveToDB(key, value);
        // 2. 更新自己的本地缓存
        localCache.put(key, value);
        // 3. 通知其他实例
        jedis.publish("cache:evict:" + key, key);
    }
}

@Component
public class VersionedCacheManager {

    private final Cache<String, VersionedEntry> localCache =
        Caffeine.newBuilder()
            .maximumSize(10_000)
            .expireAfterWrite(30, TimeUnit.SECONDS)
            .build();

    private final Jedis jedis;

    public Object get(String key) {
        VersionedEntry entry = localCache.getIfPresent(key);
        if (entry != null) {
            // 检查版本号
            Long latestVersion = jedis.hget("cache:version", key);
            if (latestVersion != null &&
                latestVersion.equals(entry.getVersion())) {
                return entry.getValue(); // 版本一致，返回缓存
            }
            // 版本不一致，失效缓存
            localCache.invalidate(key);
        }
        // 回源数据库
        Object value = loadFromDB(key);
        if (value != null) {
            Long version = jedis.hincrBy("cache:version", key, 0);
            localCache.put(key, new VersionedEntry(value, version));
        }
        return value;
    }

    public void update(String key, Object value) {
        // 1. 写数据库
        saveToDB(key, value);
        // 2. 更新版本号（原子递增）
        Long version = jedis.hincrBy("cache:version", key, 1);
        // 3. 更新本地缓存
        localCache.put(key, new VersionedEntry(value, version));
    }

    @Data
    private static class VersionedEntry {
        private final Object value;
        private final Long version;
    }
}