Redis 高级实战

# 1. 启动新节点
redis-cli --cluster add-node new_node:6379 existing_node:6379

# 2. 分配 Slot（从旧节点迁出）
redis-cli --cluster reshard old_node:6379 \
  --cluster-from node-id \
  --cluster-to new-node-id \
  --cluster-slots 4096

# 迁移过程：
# - 对每个迁移的 key：原子性的 MIGRATE 命令
# - 旧节点先锁定 key → DUMP → 新节点 RESTORE → 删除旧 key

// 方案1：布隆过滤器预过滤（推荐）
@RedisBloom
public User getUser(Long id) {
    if (!bloomFilter.mightContain("user:" + id)) {
        return null; // 一定不存在
    }
    // 继续查缓存/DB
}

// 方案2：缓存空值（简单但占用内存）
public User getUser(Long id) {
    String key = "user:" + id;
    User user = redis.get(key);
    if (user == null) {
        user = db.query(id);
        if (user == null) {
            redis.setex(key, 60, "NULL"); // 空值缓存 60s
        } else {
            redis.setex(key, 3600, user);
        }
    }
    return "NULL".equals(user) ? null : user;
}

// 方案1：互斥锁重建（排队等第一个）
public User getHotUser(Long id) {
    String key = "user:" + id;
    User user = redis.get(key);
    if (user == null) {
        String lockKey = "lock:user:" + id;
        // 获取分布式锁（SET key value NX EX 10）
        if (redis.set(lockKey, "1", "NX", "EX", 10)) {
            try {
                user = db.query(id);      // 只有一个线程查 DB
                redis.setex(key, 3600, user);
            } finally {
                redis.del(lockKey);
            }
        } else {
            Thread.sleep(100);            // 其他线程等待
            return getHotUser(id);        // 重试
        }
    }
    return user;
}

// 方案2：逻辑过期（永不过期，后台异步刷新）
@Data
class CacheItem<T> {
    T data;
    long expireTime;  // 逻辑过期时间
}

// 方案1：过期时间加随机偏移
int expireSeconds = 3600 + ThreadLocalRandom.current().nextInt(300);
redis.setex(key, expireSeconds, value);

// 方案2：多级缓存（本地缓存 + Redis）
@CaffeineCache(maxSize = 10000, expireAfterWrite = 5m)
@RedisCache(expire = 3600)
public User getUser(Long id) {
    return db.query(id);
}
// 本地缓存扛住 Redis 失效瞬间的流量洪峰

// 方案3：熔断降级
@CircuitBreaker(name = "userCache", fallbackMethod = "getUserFallback")
public User getUser(Long id) { ... }

public User getUserFallback(Long id, Exception ex) {
    return User.defaultUser(); // 返回默认值，保护数据库
}

# 发现大 Key
redis-cli --bigkeys
redis-cli --mem-keys "*" --mem-keys-samples 1000

# 使用 rdb-tools 分析
rdb -c memory dump.rdb -l largest 100 > large_keys.csv

# 大 Key 拆分示例
# 原：hash "user:10000" 有 100 万个 field → 内存爆炸
# 拆：按用户 ID 后两位分桶
#     user:10000:00 ~ user:10000:99，每个 1 万个 field

# 1. 使用高效编码
hash-max-ziplist-entries 512    # Hash 元素 < 512 用 ziplist
hash-max-ziplist-value 64       # 每个元素值 < 64B 用 ziplist
list-max-ziplist-size -2        # quicklist 节点压缩
zset-max-ziplist-entries 128    # ZSet 小数据用 ziplist

# 2. 内存淘汰策略
maxmemory-policy allkeys-lru    # 全局 LRU 淘汰（推荐）
# 备选：volatile-lru / allkeys-lfu / volatile-ttl

# 3. 内存碎片整理（Redis 4.0+）
active-defrag-yes yes
active-defrag-threshold-lower 10

# 推荐配置（混合持久化）
aof-use-rdb-preamble yes    # 先 RDB 全量 + 后续 AOF 增量
appendonly yes
appendfsync everysec        # 每秒刷盘（性能与安全的平衡）

# SET key value NX EX seconds（原子操作）
SET lock:order:1001 my_random_value NX EX 30

public boolean lock(String key, String value, int expireSeconds) {
    String result = redis.set(key, value, "NX", "EX", expireSeconds);
    return "OK".equals(result);
}

public void unlock(String key, String value) {
    // Lua 原子脚本：只释放自己持有的锁
    String script = 
        "if redis.call('get', KEYS[1]) == ARGV[1] then " +
        "  return redis.call('del', KEYS[1]) " +
        "else return 0 end";
    redis.eval(script, 1, key, value);
}

// Redisson 实现
Config config = new Config();
config.useRedLock()
    .addNodeAddress("redis://redis1:6379")
    .addNodeAddress("redis://redis2:6379")
    .addNodeAddress("redis://redis3:6379");

RedissonClient redisson = Redisson.create(config);
RLock lock = redisson.getRedLock("myLock");
lock.lock();
try {
    // 业务逻辑
} finally {
    lock.unlock();
}

# Redis Function（7.0+）
FUNCTION LOAD "#!lua name=mylib\n\nredis.register_function('myfunc', function(keys, args)\n  return redis.call('GET', keys[1])\nend)"

FCALL myfunc 1 mykey