# 1. 配置文件触发（时间+次数双条件，满足任一即触发）
save 900 1      # 900秒内至少1次写操作
save 300 10     # 300秒内至少10次写操作  
save 60 10000   # 60秒内至少10000次写操作
 
# 2. 手动触发
BGSAVE  # 后台异步执行，不阻塞主线程
SAVE    # 同步执行，会阻塞（生产环境禁用）

# 三种同步策略
appendfsync always     # 每个写命令都fsync，数据最安全（不推荐，性能差）
appendfsync everysec    # 每秒fsync一次（默认，推荐）
appendfsync no         # 由OS决定何时fsync（最快，但不安全）

# 手动触发
BGREWRITEAOF
 
# 自动触发（根据文件大小）
auto-aof-rewrite-percentage 100  # 文件大小达到上次重写的200%时触发
auto-aof-rewrite-min-size 64mb  # 文件至少达到64MB才触发

# 生产环境的最佳实践
appendonly yes                    # 开启AOF
appendfsync everysec              # 每秒同步
rdbchecksum yes                   # 开启RDB校验
dbfilename dump.rdb               # RDB文件名
dir /var/lib/redis               # 数据目录

def get_product(product_id):
    # 最前置的防护，拦截明显非法参数
    if product_id <= 0:
        return None
    # 继续业务逻辑...

def get_product(product_id):
    cache_key = f"product:{product_id}"
    
    # 先查缓存
    result = redis.get(cache_key)
    if result is not None:
        if result == "NULL":  # 空值标记
            return None
        return json.loads(result)
    
    # 查数据库
    product = db.query(f"SELECT * FROM products WHERE id={product_id}")
    
    if product:
        redis.setex(cache_key, 3600, json.dumps(product))
    else:
        # 关键：空值也要缓存，设置较短过期时间
        redis.setex(cache_key, 60, "NULL")
    
    return product

def get_product(product_id):
    cache_key = f"product:{product_id}"
    lock_key = f"lock:product:{product_id}"
    
    # 先查缓存
    result = redis.get(cache_key)
    if result:
        return json.loads(result)
    
    # 获取锁，只允许一个线程查DB
    lock = redis.pubsub("channel:{lock_key}")
    
    if redis.set(lock_key, "1", nx=True, ex=10):
        # 成功获取锁，查DB
        product = db.query(f"SELECT * FROM products WHERE id={product_id}")
        redis.setex(cache_key, 3600, json.dumps(product))
        redis.delete(lock_key)
        return product
    else:
        # 没拿到锁，短暂等待后重试
        import time
        time.sleep(0.1)
        return get_product(product_id)

# 更新策略改为主动更新，不依赖过期时间
def update_product_cache(product_id):
    # 监听数据变更事件，主动更新缓存
    product = db.query(f"SELECT * FROM products WHERE id={product_id}")
    redis.set(f"product:{product_id}", json.dumps(product))

# 不要统一设置3600秒
# 错误
redis.setex("product:{id}", 3600, data)
 
# 正确：加随机偏移
expire = 3600 + random.randint(0, 300)  # 1小时~1小时05分随机
redis.setex("product:{id}", expire, data)

主从 + 哨兵/集群
确保单节点宕机不影响服务

# 缓存雪崩时，保护数据库
def get_product_with_fallback(product_id):
    try:
        return get_product(product_id)
    except:
        # 降级：直接查DB（牺牲性能保证可用）
        return db.query(f"SELECT * FROM products WHERE id={product_id}")

┌─────────┐         ┌─────────┐
│  Master │ ──────→ │ Slave1  │  主负责写，从负责读
└─────────┘         └─────────┘
       │                   │
       └───────→ Slave2 ───┘

# 在从节点配置
replicaof 192.168.1.100 6379
 
# 或者在配置文件
replicaof <master-ip> <master-port>
replica-read-only yes  # 从节点只读

1. 全量同步：Slave连接Master，发送PSYNC命令
2. Master fork子进程生成RDB快照，发送给Slave
3. Slave加载RDB，完成全量复制
4. 增量同步：Master后续的写命令通过replication buffer发送给Slave

┌─────────┐
│ Sentinel│ ←── 监控 + 自动切换
└────┬────┘
     │
┌────┴────┐         ┌─────────┐
│  Master │ ──────→ │ Slave1  │
└─────────┘         └─────────┘
                           │
                    ┌──────┴──────┐
                    │  Sentinel   │
                    └─────────────┘

# sentinel.conf
sentinel monitor mymaster 192.168.1.100 6379 2
#                   集群名    主节点IP  端口  投票阈值
 
sentinel down-after-milliseconds mymaster 30000
# 主观下线时间（30秒没响应认为下线）
 
sentinel failover-timeout mymaster 180000
# 故障转移超时时间

1. 主观下线：Sentinel检测到Master无响应
2. 客观下线：多个Sentinel达成共识
3. 选举Leader：由Sentinel投票选出一个Leader
4. 故障转移：Leader选择一个最优Slave晋升为Master
5. 通知应用：更新配置，主从关系变更

┌─────────────────────────────────────────────────┐
│                   Redis Cluster                  │
│                                                  │
│    Slot 0-5460      Slot 5461-10922   Slot 10923-16383  │
│  ┌───────────┐    ┌───────────┐    ┌───────────┐  │
│  │  Master1  │    │  Master2  │    │  Master3  │  │
│  │  (主)     │    │  (主)     │    │  (主)     │  │
│  └─────┬─────┘    └─────┬─────┘    └─────┬─────┘  │
│        │                │                │        │
│  ┌─────┴─────┐    ┌─────┴─────┐    ┌─────┴─────┐  │
│  │  Slave1   │    │  Slave2   │    │  Slave3   │  │
│  │  (从)     │    │  (从)     │    │  (从)     │  │
│  └───────────┘    └───────────┘    └───────────┘  │
└─────────────────────────────────────────────────┘

# 计算key应该落在哪个槽
CRC16("user:10086") % 16384 = 5662  # 槽5662
 
# 查看集群槽信息
redis-cli cluster slots

# 传统取模哈希
hash(key) % node_count
 
# 问题：节点数量变化时，所有映射关系都会改变
# 例如：3个节点变成4个节点
# 几乎所有key的映射都会变化，数据大量失效

将服务器节点和数据Key都映射到一个环形哈希空间上：
 
        0
        │
        │          节点A(0°)
        │           ╱
  270°──┼─────────╱────────────────── 90°
        │        ╱       ↑
        │       ╱        │  数据Key落在这个区间
        │      ╱         │  就路由到节点A
        │     ╱          │
        │    ╱           │
  180°──┼──╱────────────┴───────────────
        │        节点B(180°)
        │
        │
        360°/0°

物理节点只有3个，但数据分布可能极不均匀：
- 节点A可能承载60%的数据
- 节点B、C各承载20%
 
解决：引入虚拟节点
 
物理节点：A、B、C
虚拟节点：A#1、A#2、A#3、B#1、B#2、B#3、C#1、C#2、C#3
 
虚拟节点映射到环上：
- A#1(30°), A#2(120°), A#3(250°)
- B#1(80°), B#2(150°), B#3(300°)
- ...
 
数据key落在节点A#2和B#1之间 → 路由到B#1（实际节点B）

# 内存 vs 磁盘延迟对比
内存访问：~100纳秒
SSD磁盘：~100微秒
机械硬盘：~10毫秒
 
内存比磁盘快 100-100000 倍

# O(1) 操作
SET key value              # 写入
GET key                    # 读取
HSET key field value       # 哈希写入
HGET key field             # 哈希读取
 
# O(k) 操作（k是参与操作的元素数量）
LPUSH key v1 v2 v3         # 列表头插，k=3
LRANGE key 0 9             # 范围查询，k=10
 
# O(n) 操作要慎用
KEYS pattern               # 全量扫描，禁止在生产使用
SCAN cursor                # 渐进式扫描，安全
FLUSHALL                   # 清空数据库，禁止随便用

单线程模型的优势：
- 没有线程切换开销
- 没有锁竞争
- 原子操作天然保证
 
IO多路复用（epoll/kqueue/select）：
- 一个线程处理多个客户端连接
- "事件发生时通知我"而非"主动轮询"
- epoll在Linux下效率极高
 
Redis 6.0多线程IO：
- 主线程处理命令执行
- 辅助线程处理socket读写、解析
- 对客户端来说仍然是单线程感知

- 内部编码优化：
  - 小列表用ziplist（内存紧凑）
  - 小集合用intset（整数数组）
  
- 文本协议（RESP）：
  - 简单，易解析
  - 序列化/反序列化开销低

T1: 线程A更新数据库：UPDATE → 新值
T2: 线程A删除缓存：DEL cache
T3: 线程B查询：cache miss → 读DB → 新值 → 写入缓存
 
结果：最终一致 ✓

T0: 初始状态，缓存miss
 
T1: 线程A查询：cache miss → 读DB → 旧值（还没写缓存）
T2: 线程B更新DB → 新值
T3: 线程B删除缓存
T4: 线程A把旧值写入缓存  ← 覆盖了！
 
结果：缓存旧值，DB新值，不一致

T1: 线程A删除缓存
T2: 线程B查询：cache miss → 读DB → 旧值
T3: 线程A更新数据库：新值
T4: 线程B把旧值写入缓存  ← 必然发生！
 
结果：必然不一致

def update_user(user_id, age):
    # 第一步：删缓存
    redis.delete(f"user:{user_id}")
    
    # 第二步：更新数据库
    db.execute(f"UPDATE users SET age={age} WHERE id={user_id}")
    
    # 第三步：延迟再删（等并发读完成）
    import time
    time.sleep(0.5)  # 等待500ms
    redis.delete(f"user:{user_id}")

# 应用层
def update_and_notify(user_id, age):
    db.execute(f"UPDATE users SET age={age} WHERE id={user_id}")
    # 发MQ消息，不同步删除
    mq.send("cache:delete", {"key": f"user:{user_id}", "retry": 0})
 
# MQ消费者
def consume(msg):
    key = msg["key"]
    if redis.delete(key):
        return  # 成功，ACK
    # 失败，重试
    if msg["retry"] < 3:
        msg["retry"] += 1
        mq.send("cache:delete", msg)
    else:
        log.error(f"删除缓存失败: {key}")

MySQL → Binlog → Canal → MQ → Worker → 删除缓存
 
优点：
- 应用层代码几乎不用改
- 数据同步是准实时的
- 删除失败有MQ兜底重试

问题：多表数据聚合后缓存
 
线程A：更新用户表年龄 age=30
线程B：更新订单表 count=100
 
如果用"更新缓存"：
T1: 线程A读用户：{age:25}，准备写缓存
T2: 线程B读订单：{count:100}，写缓存 → {age:25, count:100}
T3: 线程A写缓存：{age:30, count:旧值} ← 覆盖了订单数！
 
用"删除缓存"：
T1: 线程A更新用户表
T2: 线程A删除聚合缓存
T3: 下次查询重新聚合所有数据
 
删除比更新更安全，更简单的语义

经典比喻：100亿个商品ID去重
 
普通做法：HashSet → 几百GB内存
布隆过滤器：~1GB内存，但有0.01%误判率
 
0.01%误判率意味着：
- 100亿次查询中，约100万次误判
- 误判后去DB查一次，保证最终正确
- DB查询压力只增加0.01%，但节省99%+内存

bit数组（初始全0） + k个hash函数
 
添加"google"：
- hash1("google") = 2  → bit[2] = 1
- hash2("google") = 5  → bit[5] = 1
- hash3("google") = 8  → bit[8] = 1
 
查询"google"：
- 三个位置都是1 → 可能存在
 
查询"facebook"：
- hash1 = 2（是1）
- hash2 = 7（是1）
- hash3 = 9（是1）→ 可能存在
 
查询"orange"：
- hash1 = 2（是1）
- hash2 = 6（是0）→ 一定不存在！

假设：
- 添加"google" → bit[2,5,8] = 1
- 添加"facebook" → bit[2,7,9] = 1
 
删除"google" → bit[2,5,8] = 0
 
再查"facebook"：
- hash1 = 2 → bit[2] = 0 → 判断不存在
- 但"facebook"明明存在！
 
原因：bit[2]被google和facebook共用
删除google时把facebook的hash1位置也清掉了

# 添加元素
BF.ADD myfilter "apple"
BF.MADD myfilter "banana" "orange"
 
# 查询元素
BF.EXISTS myfilter "apple"    # 1（存在）
BF.EXISTS myfilter "watermelon" # 0（不存在）
 
# 创建自定义参数的过滤器
BF.RESERVE myfilter 0.01 1000000
# 参数：误判率1%，预计100万元素

bf = RedisBloomFilter("product:bloom")
 
def get_product(product_id):
    pid = str(product_id)
    
    # 第一层：布隆过滤器（快速判断）
    if not bf.exists(pid):
        return None  # 一定不存在，不查DB
    
    # 第二层：Redis缓存
    cache = redis.get(f"product:{pid}")
    if cache:
        return json.loads(cache)
    
    # 第三层：DB
    product = db.query(f"SELECT * FROM products WHERE id={pid}")
    
    if product:
        redis.setex(f"product:{pid}", 3600, json.dumps(product))
        bf.add(pid)  # 同步添加
    else:
        # 关键：DB没有也要加入过滤器
        bf.add(pid)
    
    return product

# 错误！不是原子操作
redis.setnx(lock_key, "1")
redis.expire(lock_key, 10)
# 如果setnx成功后expire执行前崩溃，锁永不释放

def acquire(key, expire_seconds=10):
    lock_key = f"lock:{key}"
    # SET + NX + EX 是原子操作
    result = redis.set(lock_key, "1", nx=True, ex=expire_seconds)
    return result is not None
 
def release(key):
    redis.delete(f"lock:{key}")

import uuid
 
def acquire(key, expire_seconds=10):
    lock_key = f"lock:{key}"
    token = str(uuid.uuid4())
    
    result = redis.set(lock_key, token, nx=True, ex=expire_seconds)
    if result:
        return token
    return None
 
def release(key, token):
    # Lua脚本保证原子性：检查token匹配后再删除
    lua = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """
    redis.eval(lua, 1, f"lock:{key}", token)

// 看门狗原理：
// 1. 锁默认leaseTime=30秒
// 2. 每隔10秒，自动续期30秒
// 3. 只要持有锁的进程还活着，锁就不会过期
// 4. 进程崩溃 → 不续期 → 锁30秒后自动释放
 
RLock lock = redisson.getLock("order:12345");
try {
    // tryLock(等待时间, 锁定时间, 时间单位)
    // 如果leaseTime=-1，使用看门狗机制
    boolean acquired = lock.tryLock(10, 30, TimeUnit.SECONDS);
    
    if (acquired) {
        // 业务逻辑
    }
} finally {
    if (lock.isHeldByCurrentThread()) {
        lock.unlock();
    }
}

问题：
1. 客户端A获取master的锁
2. master挂掉，slave晋升为master
3. 新master没有锁信息
4. 客户端B也能获取锁成功
 
解决方案：RedLock算法
- 5个独立Redis节点
- 超过半数节点获取成功才算成功
- 即使2个节点故障，锁仍然有效

RLock lock = redisson.getLock("stock:" + productId);
try {
    lock.lock(30, TimeUnit.SECONDS);
    
    int stock = Integer.parseInt(redis.get("stock:" + productId));
    if (stock < quantity) {
        throw new Exception("库存不足");
    }
    redis.decrby("stock:" + productId, quantity);
} finally {
    lock.unlock();
}

RLock lock = redisson.getLock("task:dailyReport");
if (lock.tryLock()) {
    try {
        generateDailyReport();
    } finally {
        lock.unlock();
    }
}
// 其他节点跳过执行

# 快速扫描
redis-cli --bigkeys
 
# 精确排查
redis-cli --scan | head -1000 | while read key; do
    echo "$(redis-cli MEMORY USAGE "$key") $key"
done | sort -rn | head -20
 
# 查看内存碎片率
redis-cli INFO memory | grep mem_fragmentation_ratio
# 超过1.5需要关注，超过1.8需要重启修复

# 原来：一个大Hash存用户画像
user:10086:profile → 100万个field
 
# 优化后：按业务维度拆分
user:10086:basic      → 基本信息（100字段）
user:10086:behavior   → 行为数据（1000字段）
user:10086:social     → 社交数据（500字段）

# 原来：一个关注列表1百万粉丝
user:10086:followers
 
# 优化后：分桶
user:10086:followers:0   → 0-9999
user:10086:followers:1   → 10000-19999
user:10086:followers:2   → 20000-29999

# DEL会阻塞，用UNLINK异步删除
redis.unlink("large:key")