4. 热点Key的发现手段二：基于客户端统计（Jedis/Redisson）的本地计数方案

上一节我们介绍了通过代理层（如Twemproxy、Codis）进行全局热点统计的方案。本节将聚焦于另一种轻量级、无侵入的发现手段——客户端本地计数方案。该方案通过在业务应用（Jedis/Redisson客户端）内部嵌入计数器，实时统计每个Key的访问频率，从而在客户端侧直接识别潜在的热点Key。

4.1 方案原理与核心思想

客户端本地计数方案的核心思想是：将统计逻辑下沉到业务应用进程内，利用本地内存（如ConcurrentHashMap、LongAdder）对每次Redis操作的目标Key进行计数。当某个Key的本地访问频率超过预设阈值时，触发告警或上报逻辑。

下图展示了客户端本地计数方案的整体架构与数据流：

4.2 基于Jedis的本地计数实现

在Jedis客户端中，我们可以通过包装JedisPool或直接拦截Jedis命令执行来实现计数。以下是一个基于装饰器模式的轻量级实现示例：

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

public class HotKeyJedisWrapper {
    // 本地计数器：Key -> 访问次数
    private final ConcurrentHashMap<String, LongAdder> counterMap = new ConcurrentHashMap<>();
    // 热点阈值：每秒超过100次视为热点
    private static final long HOT_THRESHOLD = 100;
    // 统计窗口（毫秒）
    private static final long WINDOW_MS = 1000;

    private final JedisPool jedisPool;

    public HotKeyJedisWrapper(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
        // 启动定时清理任务，每秒重置计数器
        startCleanupTask();
    }

    public String get(String key) {
        // 1. 本地计数
        incrementCounter(key);
        // 2. 执行实际Redis操作
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.get(key);
        }
    }

    public void set(String key, String value) {
        incrementCounter(key);
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.set(key, value);
        }
    }

    private void incrementCounter(String key) {
        LongAdder counter = counterMap.computeIfAbsent(key, k -> new LongAdder());
        counter.increment();
        long currentCount = counter.sum();
        if (currentCount > HOT_THRESHOLD) {
            // 触发热点告警
            System.err.println("[热点告警] Key: " + key + " 当前计数: " + currentCount);
            // 可在此处上报到监控中心或写入日志
        }
    }

    private void startCleanupTask() {
        // 使用ScheduledExecutorService定期重置计数器
        // 实际生产建议使用ScheduledThreadPoolExecutor
        new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(WINDOW_MS);
                    counterMap.clear(); // 简单重置，生产环境建议使用滑动窗口
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
}

4.3 基于Redisson的本地计数实现

Redisson作为高级客户端，提供了更丰富的分布式数据结构。我们可以利用Redisson的RLocalCachedMap或自定义拦截器来实现本地计数。以下是一个基于Redisson CommandInterceptor的优雅实现：

import org.redisson.api.RedissonClient;
import org.redisson.api.RMap;
import org.redisson.client.protocol.decoder.ListScanResult;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.command.CommandInterceptor;
import org.redisson.misc.RPromise;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

public class HotKeyRedissonInterceptor implements CommandInterceptor {

    private final ConcurrentHashMap<String, LongAdder> localCounter = new ConcurrentHashMap<>();
    private static final long HOT_THRESHOLD = 200; // Redisson场景阈值可适当提高
    private static final long WINDOW_MS = 1000;

    public HotKeyRedissonInterceptor() {
        startCleanupTask();
    }

    @Override
    public <T> RPromise<T> interceptCommand(CommandAsyncExecutor executor,
                                              RedisCommand<T> command,
                                              Object... params) {
        // 提取Key参数（通常第一个参数是Key）
        if (params != null && params.length > 0 && params[0] instanceof String) {
            String key = (String) params[0];
            // 只对读写命令进行计数
            if (isReadWriteCommand(command)) {
                incrementCounter(key);
            }
        }
        // 继续执行原始命令
        return executor.executeAsync(command, params);
    }

    private boolean isReadWriteCommand(RedisCommand<?> command) {
        String cmdName = command.getName().toUpperCase();
        return cmdName.equals("GET") || cmdName.equals("SET") ||
               cmdName.equals("HGET") || cmdName.equals("HSET") ||
               cmdName.equals("ZADD") || cmdName.equals("ZRANGE");
    }

    private void incrementCounter(String key) {
        LongAdder counter = localCounter.computeIfAbsent(key, k -> new LongAdder());
        counter.increment();
        long count = counter.sum();
        if (count > HOT_THRESHOLD) {
            // 上报热点
            reportHotKey(key, count);
        }
    }

    private void reportHotKey(String key, long count) {
        // 实际生产：写入日志、发送到Kafka或调用告警API
        System.out.println("[Redisson热点] Key: " + key + " 计数: " + count + "/s");
    }

    private void startCleanupTask() {
        // 使用ScheduledExecutorService实现
    }
}

4.4 方案对比：Jedis vs Redisson

4.5 本地计数方案的局限性

• 数据分散性： 每个客户端实例独立计数，无法感知全局热点。例如，100个客户端每个计数80次/秒，本地均未达阈值，但全局已达8000次/秒，实为热点。
• 内存占用： 每个客户端维护一个计数器Map，Key数量多时（如百万级）会占用大量堆内存，可能引发GC压力。
• 阈值难统一： 不同客户端流量不均，固定阈值难以适配所有实例，动态阈值实现复杂。
• 重启丢失： 应用重启后计数器清零，热点发现存在盲区。

下图直观展示了本地计数方案的“视野盲区”问题：

4.6 最佳实践与优化建议

1. 滑动窗口替代固定窗口： 使用环形数组或时间轮实现滑动窗口计数，避免窗口边界毛刺。
2. 采样统计： 不必100%计数，可采用概率采样（如1/10采样），降低CPU和内存开销。
3. 分级阈值： 设置告警阈值和熔断阈值两级，告警阈值用于通知，熔断阈值触发本地降级（如直接返回缓存）。
4. 结合代理层方案： 客户端本地计数作为第一道防线，代理层全局统计作为第二道防线，形成互补。
5. 热点Key本地缓存： 一旦识别出热点Key，可在客户端侧增加一级本地缓存（如Caffeine），减少对Redis的冲击。