141｜热 Key 与大 Key：识别、拆分、复制、限流和删除策略

# 使用 redis-cli 的 --hotkeys 参数（需要 Redis 4.0+）
redis-cli --hotkeys -a your_password

# 输出示例：
# Hot Keys found:
# [mykey1] has 12345 requests in 1 second
# [mykey2] has 9876 requests in 1 second

redis-cli MONITOR | grep "GET\|HGET" | awk '{print $4}' | sort | uniq -c | sort -nr | head -20

// 使用 Jedis/Lettuce 的拦截器统计热 Key
public class HotKeyInterceptor implements CommandInterceptor {
    private ConcurrentHashMap<String, AtomicLong> keyCounter = new ConcurrentHashMap<>();
    private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public HotKeyInterceptor() {
        // 每秒上报统计
        scheduler.scheduleAtFixedRate(() -> {
            keyCounter.forEach((key, count) -> {
                if (count.get() > 1000) { // 阈值
                    log.warn("Hot key detected: {}, QPS: {}", key, count.get());
                }
            });
            keyCounter.clear();
        }, 1, 1, TimeUnit.SECONDS);
    }
}

CONFIG SET audit-log-enabled yes
CONFIG SET audit-log-max-age 3600

// 使用 Caffeine 实现两级缓存
public class TwoLevelCache {
    private Cache<String, Object> localCache = Caffeine.newBuilder()
        .maximumSize(1000)
        .expireAfterWrite(3, TimeUnit.SECONDS)  // 本地缓存过期时间短
        .build();

    private RedisTemplate<String, Object> redis;

    public Object get(String key) {
        // 一级缓存：本地
        Object value = localCache.getIfPresent(key);
        if (value != null) return value;

        // 二级缓存：Redis
        value = redis.opsForValue().get(key);
        if (value != null) {
            localCache.put(key, value);
        }
        return value;
    }
}

public class HotKeyReplicator {
    // key_副本序号 作为实际 Key
    private static final int REPLICA_COUNT = 10;

    public Object get(String key) {
        // 随机选择副本
        int replicaIndex = ThreadLocalRandom.current().nextInt(REPLICA_COUNT);
        String replicaKey = key + "_rep_" + replicaIndex;
        return redis.get(replicaKey);
    }

    public void set(String key, Object value) {
        // 写操作：更新所有副本
        for (int i = 0; i < REPLICA_COUNT; i++) {
            redis.set(key + "_rep_" + i, value, 10, TimeUnit.SECONDS);
        }
    }
}

// 将 total_stock 拆分为 N 个分片
public class StockSplitter {
    private static final int SEGMENT_COUNT = 20;

    public Long deduct(String productId, String userId) {
        // 按用户 ID 哈希路由到不同的库存分片
        int segmentIndex = Math.abs(userId.hashCode()) % SEGMENT_COUNT;
        String segmentKey = "stock:" + productId + ":" + segmentIndex;

        // 使用 Lua 脚本保证原子性扣减
        String script = "local stock = redis.call('get', KEYS[1]) " +
                        "if stock and tonumber(stock) > 0 then " +
                        "    return redis.call('decr', KEYS[1]) " +
                        "else return -1 end";
        return redis.eval(script, List.of(segmentKey));
    }
}

// 使用令牌桶 + Redis 实现分布式限流
public class HotKeyRateLimiter {
    public boolean isAllowed(String key, int maxQps) {
        String luaScript =
            "local key = KEYS[1] " +
            "local limit = tonumber(ARGV[1]) " +
            "local current = redis.call('get', key) " +
            "if current and tonumber(current) >= limit then " +
            "    return 0 " +
            "else " +
            "    redis.call('incr', key) " +
            "    redis.call('expire', key, 1) " +
            "    return 1 " +
            "end";
        return redis.eval(luaScript, List.of("rate_limit:" + key), String.valueOf(maxQps)) == 1L;
    }
}

# 删除一个 500 万元素的 Set
DEL big_set_key
# 这个操作可能阻塞 Redis 主线程 1-2 秒！

redis-cli --bigkeys -a your_password

# 输出示例：
# Biggest string found: "user:session:123456" has 5242880 bytes
# Biggest hash found: "product:detail:789" has 25000 fields
# Biggest list found: "msg_queue:order" has 100000 items

# 精确查询单个 Key 的内存占用
MEMORY USAGE mykey
# 返回：1234567（字节）

# 抽样估算
MEMORY USAGE mykey SAMPLES 1000

# 安装
pip install rdbtools python-lzf

# 分析 RDB 文件，生成 CSV 报告
rdb -c memory /var/redis/dump.rdb > memory_report.csv

# 按内存占用排序
sort -t',' -k4 -n memory_report.csv | tail -20

# 定期扫描大 Key 的脚本
def scan_big_keys(redis_client, threshold=10240):
    cursor = 0

    big_keys = [ ]

    while True:
        cursor, keys = redis_client.scan(cursor, count=1000)
        for key in keys:
            key_type = redis_client.type(key).decode()
            if key_type == 'string':
                size = redis_client.strlen(key)
            elif key_type == 'hash':
                size = redis_client.hlen(key)
            elif key_type == 'list':
                size = redis_client.llen(key)
            elif key_type == 'set':
                size = redis_client.scard(key)
            elif key_type == 'zset':
                size = redis_client.zcard(key)
            if size > threshold:
                big_keys.append((key, key_type, size))
        if cursor == 0:
            break
    return big_keys

// 按字段哈希拆分
public class HashSplitter {
    private static final int SHARD_COUNT = 100;

    public void hset(String key, String field, String value) {
        int shard = Math.abs(field.hashCode()) % SHARD_COUNT;
        String shardKey = key + ":" + shard;
        redis.hset(shardKey, field, value);
    }

    public String hget(String key, String field) {
        int shard = Math.abs(field.hashCode()) % SHARD_COUNT;
        String shardKey = key + ":" + shard;
        return redis.hget(shardKey, field);
    }

    // 获取所有字段需要遍历所有分片
    public Map<String, String> hgetAll(String key) {
        Map<String, String> result = new HashMap<>();
        for (int i = 0; i < SHARD_COUNT; i++) {
            result.putAll(redis.hgetAll(key + ":" + i));
        }
        return result;
    }
}

# 存储用户 ID 列表 → 用 Set 还是 List？
# 如果不需要去重和交集操作，用 List 比 Set 内存占用更小
# Redis 的 Set 使用哈希表实现，内存开销较大

# 配置压缩编码的阈值
hash-max-listpack-entries 512
hash-max-listpack-value 64

# UNLINK 是异步删除，不会阻塞主线程
UNLINK big_set_key

// 使用 SCAN + 分批删除（适用于 Hash/Set/ZSet）
public void batchDelete(RedisTemplate redis, String key) {
    // 1. 先重命名，防止新的写入
    String tmpKey = key + ":to_delete:" + System.currentTimeMillis();
    redis.rename(key, tmpKey);

    // 2. 分批删除
    ScanOptions options = ScanOptions.scanOptions().count(100).build();
    Cursor<Map.Entry<Object, Object>> cursor = redis.opsForHash().scan(tmpKey, options);
    while (cursor.hasNext()) {
        Map.Entry<Object, Object> entry = cursor.next();
        redis.opsForHash().delete(tmpKey, entry.getKey());
        // 每次删除后短暂休眠，让出 CPU
        Thread.sleep(10);
    }

    // 3. 最后删除空 Key
    redis.delete(tmpKey);
}

public class CompressedRedisTemplate {
    public void set(String key, String value) {

        byte[ ] compressed = gzipCompress(value.getBytes(StandardCharsets.UTF_8));

        redis.opsForValue().set(key, compressed);
    }

    public String get(String key) {

        byte[ ] compressed = redis.opsForValue().get(key);

        return new String(gzipDecompress(compressed), StandardCharsets.UTF_8);
    }
}

# 避免大量 Key 在同一时间过期，加随机偏移
EXPIRE mykey 3600 + RANDOM(0, 300)

发现 → 分析 → 治理 → 验证 → 闭环
  │      │      │      │      │
  │      │      │      │      └─ 沉淀规则，更新监控
  │      │      │      └─ 验证治理效果，确认无副作用
  │      │      └─ 选择方案，执行治理
  │      └─ 分析根因，判断类型
  └─ 通过监控、扫描发现异常 Key

// 热 Key 自动发现与治理框架
@Component
public class HotKeyGovernor {
    // 1. 发现层：基于客户端 SDK 统计
    private HotKeyDetector detector;

    // 2. 分析层：判断热 Key 类型和严重程度
    private HotKeyAnalyzer analyzer;

    // 3. 治理层：自动执行治理策略
    private HotKeyMitigator mitigator;

    // 4. 验证层：确认治理效果
    private HotKeyValidator validator;

    @Scheduled(fixedDelay = 10000)
    public void govern() {
        // 发现热 Key
        List<HotKeyReport> reports = detector.detect();
        for (HotKeyReport report : reports) {
            // 分析
            HotKeyType type = analyzer.analyze(report);

            // 治理
            switch (type) {
                case READ_ONLY:
                    // 读多写少 → 本地缓存 + 多副本
                    mitigator.applyLocalCache(report.getKey());
                    break;
                case READ_WRITE:
                    // 读写频繁 → Key 拆分
                    mitigator.splitKey(report.getKey());
                    break;
                case BURST:
                    // 突发流量 → 限流
                    mitigator.applyRateLimit(report.getKey());
                    break;
            }

            // 验证
            validator.validate(report.getKey());
        }
    }
}

# 绝对不要在生产环境执行！
KEYS *  # 这个命令会阻塞 Redis 主线程

// Redis 4.0 之前，delCommand 直接调用 dbAsyncDelete 或 dbSyncDelete
// 默认是同步删除
void delCommand(client *c) {
    int deleted = 0, j;
    for (j = 1; j < c->argc; j++) {
        // 同步删除，直接释放内存
        if (dbSyncDelete(c->db, c->argv[j])) {
            // ...
            deleted++;
        }
    }
    addReplyLongLong(c, deleted);
}

// Redis 4.0+ 引入 lazyfree 机制
// dbAsyncDelete 将内存释放操作交给后台线程
int dbAsyncDelete(redisDb *db, robj *key) {
    // 从键空间删除
    if (dictSize(db->expires) > 0) dictDelete(db->expires, key->ptr);
    dictEntry *de = dictUnlink(db->dict, key->ptr);
    if (de) {
        robj *val = dictGetVal(de);
        // 计算释放成本
        size_t free_effort = lazyfreeGetFreeEffort(key, val);
        // 如果释放成本 > 64 且共享计数为 1，则异步释放
        if (free_effort > LAZYFREE_THRESHOLD && val->refcount == 1) {
            // 放入后台线程队列
            atomicIncr(lazyfree_objects, 1);
            bioCreateLazyFreeJob(lazyfreeFreeObject, 1, val);
            dictSetVal(db->dict, de, NULL);
        }
    }
    // ...
}

# 最佳实践：评估后再决定
# 1. 先用 MEMORY USAGE 查看大小
MEMORY USAGE mykey
# 2. 如果 > 10MB，使用 UNLINK
UNLINK mykey
# 3. 如果 > 100MB，使用分批删除 + RENAME