一致性哈希在生产中的坑：虚拟节点数量、数据倾斜、热点问题

适读人群：中高级Java工程师 | 阅读时长：约18分钟 | 技术栈：Spring Boot 3.x、Redis Cluster、Caffeine

---

开篇故事

2021年我们自研了一个分布式缓存路由层，底层是 5 个 Redis 节点，用一致性哈希做数据路由。上线后第一周一切正常，第二周开始，一号节点的 CPU 持续高于其他节点 30%，内存也多占了约 20%。

起初以为是偶然，观察了三天发现稳定复现，而且随着时间推移，倾斜越来越严重。最终排查发现：我们的虚拟节点数量设置的是 10 个，加上物理节点的哈希分布不均匀，一号节点实际承载了约 30% 的数据，而最轻的四号节点只承载了 14%。

更糟的是，我们有几个热门商品（ID 特别集中的几个范围），恰好都哈希到了一号节点，导致热点叠加数据倾斜，一号节点成了整个缓存层的瓶颈。

这次事故让我把一致性哈希研究了个透彻。

---

一、核心问题分析

普通哈希的问题

普通取模哈希（hash(key) % N）在节点数 N 变化时，几乎所有 key 都需要重新映射。假设有 3 个节点，数据均匀分布，此时加一个节点变成 4 个节点，每个 key 的映射规则从 %3 变成 %4，大约 75% 的 key 需要重新分配，意味着要迁移约 75% 的数据。对于大型缓存系统，这是灾难性的。

一致性哈希的基本原理

一致性哈希将所有节点和 key 都映射到同一个哈希环（0 到 2^32-1 的整数范围）上，每个 key 顺时针找到第一个节点就是它的归属节点。

好处：加减节点时，只有该节点相邻区间的 key 需要迁移，理想情况下只影响 1/N 的数据。

问题：当节点数量少时，哈希环上的节点分布可能严重不均匀，导致数据倾斜。

虚拟节点的作用和选择

虚拟节点是解决分布不均的标准方案：每个物理节点在哈希环上对应多个虚拟节点，虚拟节点越多，分布越均匀。

但虚拟节点数量如何选择？这里没有标准答案，取决于节点数量和可以接受的内存开销。一般建议：

• 节点数量 < 10：虚拟节点 150-200 个
• 节点数量 10-50：虚拟节点 100-150 个
• 节点数量 > 50：虚拟节点 50-100 个

我们出事的原因就是虚拟节点只设了 10 个，太少了，分布严重不均匀。

---

二、原理深度解析

一致性哈希环的数学分析

只有 3 个节点时，哈希值可能这样分布：

• 节点A：hash = 100，负责区间 (800, 100]（含环绕）约占 37.5%
• 节点B：hash = 500，负责区间 (100, 500] 约占 12.5%
• 节点C：hash = 800，负责区间 (500, 800] 约占 18.75%

分布极其不均匀，A 节点承载了 B 节点 3 倍的数据量。引入虚拟节点后，每个物理节点在环上有多个位置，整体分布趋向均匀。

数据倾斜的量化分析

// 模拟不同虚拟节点数量下的数据分布均匀性
public class ConsistentHashAnalyzer {

    public static void analyzeDistribution(int nodeCount, int virtualNodes, int keyCount) {
        ConsistentHashRing ring = new ConsistentHashRing(virtualNodes);
        for (int i = 0; i < nodeCount; i++) {
            ring.addNode("node" + i);
        }

        Map<String, Integer> distribution = new HashMap<>();
        for (int i = 0; i < keyCount; i++) {
            String node = ring.getNode("key" + i);
            distribution.merge(node, 1, Integer::sum);
        }

        // 计算标准差，衡量分布均匀性
        double avg = (double) keyCount / nodeCount;
        double variance = distribution.values().stream()
            .mapToDouble(v -> Math.pow(v - avg, 2))
            .average().orElse(0);
        double stdDev = Math.sqrt(variance);
        double cv = stdDev / avg * 100; // 变异系数，越小越均匀

        System.out.printf("节点数=%d, 虚拟节点=%d, 变异系数=%.2f%%%n",
            nodeCount, virtualNodes, cv);
        distribution.forEach((node, count) ->
            System.out.printf("  %s: %d (%.1f%%)%n", node, count, count * 100.0 / keyCount));
    }
}

实际测试结果（5个节点，1万个key）：

• 虚拟节点=10：变异系数约 45%，各节点数据量差异巨大
• 虚拟节点=50：变异系数约 18%，明显改善
• 虚拟节点=150：变异系数约 8%，比较均匀
• 虚拟节点=500：变异系数约 4%，很均匀，但内存开销大

---

三、完整代码实现

一致性哈希核心实现

@Slf4j
public class ConsistentHashRing<T> {

    // 哈希环：有序映射，key 是哈希值，value 是节点
    private final TreeMap<Long, T> ring = new TreeMap<>();

    // 每个物理节点的虚拟节点数量
    private final int virtualNodes;

    // 哈希函数（使用 MurmurHash3，比 MD5 快且分布均匀）
    private final HashFunction hashFunction;

    public ConsistentHashRing(int virtualNodes) {
        this.virtualNodes = virtualNodes;
        this.hashFunction = Hashing.murmur3_128();
    }

    /**
     * 添加节点
     */
    public void addNode(T node) {
        for (int i = 0; i < virtualNodes; i++) {
            String virtualNodeKey = node.toString() + "#VN" + i;
            long hash = hash(virtualNodeKey);
            ring.put(hash, node);
        }
        log.debug("添加节点 {}，当前虚拟节点总数={}", node, ring.size());
    }

    /**
     * 移除节点
     */
    public void removeNode(T node) {
        for (int i = 0; i < virtualNodes; i++) {
            String virtualNodeKey = node.toString() + "#VN" + i;
            long hash = hash(virtualNodeKey);
            ring.remove(hash);
        }
        log.debug("移除节点 {}，当前虚拟节点总数={}", node, ring.size());
    }

    /**
     * 获取 key 对应的节点
     */
    public T getNode(String key) {
        if (ring.isEmpty()) {
            throw new IllegalStateException("哈希环为空，没有可用节点");
        }
        long hash = hash(key);
        // ceilingEntry: 找到第一个 >= hash 的虚拟节点
        Map.Entry<Long, T> entry = ring.ceilingEntry(hash);
        if (entry == null) {
            // hash 比所有虚拟节点都大，环绕到第一个节点
            entry = ring.firstEntry();
        }
        return entry.getValue();
    }

    /**
     * 获取 key 对应的 N 个不同物理节点（用于副本路由）
     */
    public List<T> getNodes(String key, int replicaCount) {
        if (ring.isEmpty()) {
            return Collections.emptyList();
        }

        Set<T> nodes = new LinkedHashSet<>();
        long hash = hash(key);

        // 从 hash 位置开始顺时针找 replicaCount 个不同物理节点
        NavigableMap<Long, T> tailMap = ring.tailMap(hash);
        for (T node : tailMap.values()) {
            nodes.add(node);
            if (nodes.size() == replicaCount) {
                return new ArrayList<>(nodes);
            }
        }

        // 如果尾部不够，从头部继续找
        for (T node : ring.values()) {
            nodes.add(node);
            if (nodes.size() == replicaCount) {
                return new ArrayList<>(nodes);
            }
        }

        return new ArrayList<>(nodes);
    }

    /**
     * 获取当前数据分布情况
     */
    public Map<T, Double> getDistribution() {
        if (ring.isEmpty()) {
            return Collections.emptyMap();
        }

        Map<T, Long> nodeRangeMap = new HashMap<>();
        long prev = ring.lastKey(); // 环的起点是最后一个key+1

        for (Map.Entry<Long, T> entry : ring.entrySet()) {
            long curr = entry.getKey();
            long range = curr >= prev ? curr - prev : (Long.MAX_VALUE - prev) + curr;
            nodeRangeMap.merge(entry.getValue(), range, Long::sum);
            prev = curr;
        }

        long total = nodeRangeMap.values().stream().mapToLong(Long::longValue).sum();
        Map<T, Double> distribution = new HashMap<>();
        nodeRangeMap.forEach((node, range) ->
            distribution.put(node, (double) range / total * 100));

        return distribution;
    }

    private long hash(String key) {
        return hashFunction.hashString(key, StandardCharsets.UTF_8).asLong()
               & Long.MAX_VALUE; // 确保为正数
    }
}

热点检测与处理

@Service
@Slf4j
public class HotKeyDetector {

    // 使用 Caffeine 做本地滑动窗口计数
    private final Cache<String, AtomicLong> hotKeyCounter = Caffeine.newBuilder()
        .expireAfterWrite(Duration.ofSeconds(10))
        .maximumSize(10_000)
        .build();

    // 热点阈值：10秒内访问超过1000次认为是热点
    private static final long HOT_KEY_THRESHOLD = 1000;

    // 本地热点缓存（避免热点key频繁路由到远端）
    private final Cache<String, Object> localHotCache = Caffeine.newBuilder()
        .expireAfterWrite(Duration.ofSeconds(5)) // 热点本地缓存5秒
        .maximumSize(1000)
        .build();

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private ConsistentHashRing<String> hashRing;

    /**
     * 获取缓存值，带热点检测
     */
    public String get(String key) {
        // 先查本地热点缓存
        Object localValue = localHotCache.getIfPresent(key);
        if (localValue != null) {
            return (String) localValue;
        }

        // 访问计数
        AtomicLong counter = hotKeyCounter.get(key, k -> new AtomicLong(0));
        long count = counter.incrementAndGet();

        // 检测热点
        if (count > HOT_KEY_THRESHOLD) {
            log.warn("热点 key 检测：key={}, 10秒内访问={}次", key, count);
        }

        // 路由到对应节点
        String node = hashRing.getNode(key);
        String value = getFromNode(node, key);

        // 如果是热点，放入本地缓存
        if (count > HOT_KEY_THRESHOLD && value != null) {
            localHotCache.put(key, value);
            log.info("热点 key 放入本地缓存：key={}", key);
        }

        return value;
    }

    /**
     * 热点 key 写入时，同步更新本地热点缓存
     */
    public void set(String key, String value, Duration ttl) {
        String node = hashRing.getNode(key);
        setToNode(node, key, value, ttl);

        // 如果本地有热点缓存，同步更新
        if (localHotCache.getIfPresent(key) != null) {
            localHotCache.put(key, value);
        }
    }

    private String getFromNode(String node, String key) {
        // 根据 node 路由到对应 Redis 实例
        return redisTemplate.opsForValue().get(key);
    }

    private void setToNode(String node, String key, String value, Duration ttl) {
        redisTemplate.opsForValue().set(key, value, ttl);
    }
}

节点扩缩容时的数据迁移

@Service
@Slf4j
public class HashRingMigrationService {

    @Autowired
    private ConsistentHashRing<RedisNode> hashRing;

    /**
     * 添加新节点，迁移受影响的 key
     * 只迁移在新节点到其前驱节点之间的 key（1/N 的数据量）
     */
    public MigrationResult addNodeWithMigration(RedisNode newNode, List<String> allKeys) {
        // 记录迁移前每个 key 的节点
        Map<String, RedisNode> beforeMapping = new HashMap<>();
        allKeys.forEach(key -> beforeMapping.put(key, hashRing.getNode(key)));

        // 加入新节点
        hashRing.addNode(newNode);

        // 找出需要迁移的 key（路由变化的 key）
        List<String> toMigrate = new ArrayList<>();
        allKeys.forEach(key -> {
            RedisNode newMapping = hashRing.getNode(key);
            if (!newMapping.equals(beforeMapping.get(key))) {
                toMigrate.add(key);
            }
        });

        log.info("节点扩容，需要迁移 key 数量={}/{}", toMigrate.size(), allKeys.size());

        // 执行迁移
        int migratedCount = 0;
        for (String key : toMigrate) {
            RedisNode source = beforeMapping.get(key);
            RedisNode target = hashRing.getNode(key);
            if (migrateKey(key, source, target)) {
                migratedCount++;
            }
        }

        return MigrationResult.builder()
            .totalKeys(allKeys.size())
            .migratedKeys(migratedCount)
            .migrationRatio((double) migratedCount / allKeys.size())
            .build();
    }

    private boolean migrateKey(String key, RedisNode source, RedisNode target) {
        try {
            // 从源节点读取数据
            String value = source.get(key);
            Long ttl = source.ttl(key);
            if (value == null) {
                return false;
            }
            // 写入目标节点
            if (ttl > 0) {
                target.setex(key, value, ttl);
            } else {
                target.set(key, value);
            }
            // 删除源节点数据（可选，等 TTL 自然过期也可以）
            // source.delete(key);
            return true;
        } catch (Exception e) {
            log.error("key 迁移失败：key={}", key, e);
            return false;
        }
    }
}

---

四、生产调优与配置

虚拟节点数量的权衡

选择虚拟节点数量要平衡以下因素：

• 内存开销：每个虚拟节点需要在 TreeMap 中存储一个 Long -> Node 的映射，5个节点+150个虚拟节点 = 750条记录，内存很小。
• 均匀性：虚拟节点越多，哈希环分布越均匀，但需要实际测试验证。
• 路由时间：TreeMap 的查询是 O(log N)，N 是虚拟节点总数，影响很小。

生产建议：至少 150 个虚拟节点，可以用上面的分析器代码实测你的业务数据分布。

热点处理策略

处理热点的标准三板斧：

本地缓存（如上代码所示）：热点 key 在每个实例的本地缓存一份，直接内存读取，不走网络。缺点是写入时需要更新所有实例的本地缓存（或等 TTL 过期）。

key 打散：对热点 key 添加随机后缀，映射到多个物理节点：hot_product:1234:0、hot_product:1234:1、hot_product:1234:2，读取时随机选一个，写入时同步所有后缀。

专用热点节点：检测到热点后，动态将该 key 路由到专门的热点集群（更高配置的 Redis），与普通数据隔离。

---

五、踩坑实录

坑一：虚拟节点数量不足导致数据倾斜（开篇故事复盘）

虚拟节点从 10 改为 150 后，我用分析器测试了一下分布：5个节点时，变异系数从 45% 降到了 8%，最大节点数据量从 30% 降到了 23%。一号节点的 CPU 从高出 30% 回落到与其他节点持平。

坑二：MurmurHash 和 MD5 的分布差异

最初用 MD5 做哈希函数，后来切换到 MurmurHash3。测试发现，相同虚拟节点数下，MurmurHash3 的变异系数比 MD5 低约 3-5 个百分点，分布更均匀，而且 MurmurHash3 的计算速度是 MD5 的 5-10 倍（没有密码学安全性要求，不需要 MD5）。

坑三：节点下线时的数据丢失

某次我们下线了一个 Redis 节点（计划内维护），从哈希环中移除了这个节点，但忘记提前迁移数据。节点下线后，原来路由到这个节点的 key 全部 miss，应用层回源查 DB，DB 被打爆，QPS 从 3000 飙到 15000，连接数耗尽，影响了约 15 分钟。

正确做法：节点下线前，先执行数据迁移（把该节点的 key 迁移到其他节点），迁移完成后再从哈希环中移除并下线物理节点。

---

六、总结

一致性哈希在生产中的关键控制点：

一、虚拟节点数量不低于 150（物理节点少于 50 个时），用分析器实测均匀性。 二、哈希函数用 MurmurHash3，比 MD5 快且分布更均匀。 三、热点检测要前置，不要等 Redis 节点被打爆再处理，用滑动窗口计数提前识别。 四、节点扩缩容一定要先迁移数据，不能靠 key miss 回源兜底。 五、定期检测分布均匀性，随着业务 key 的分布变化，定期重新评估虚拟节点设置。